diff --git a/agentkit/platform/configuration.py b/agentkit/platform/configuration.py index 5366962..7070d43 100644 --- a/agentkit/platform/configuration.py +++ b/agentkit/platform/configuration.py @@ -164,6 +164,7 @@ def get_service_credentials(self, service_key: str) -> Credentials: 3. Global Env Vars 4. Global Config File 5. VeFaaS IAM + 6. .env file in current working directory (fallback) """ # 1. Explicit if self._ak and self._sk: @@ -189,11 +190,76 @@ def get_service_credentials(self, service_key: str) -> Credentials: if creds := self._get_credential_from_vefaas_iam(): return creds + # 6. .env file fallback (Current Working Directory) + if creds := self._get_dotenv_credentials(service_key): + return creds + raise ValueError( - f"Volcengine credentials not found (Service: {service_key}). Please set environment variables VOLCENGINE_ACCESS_KEY and " - "VOLCENGINE_SECRET_KEY, or configure in global config file ~/.agentkit/config.yaml." + "\n".join( + [ + f"Volcengine credentials not found (Service: {service_key}).", + "Recommended (global, set once):", + " agentkit config --global --set volcengine.access_key=YOUR_ACCESS_KEY", + " agentkit config --global --set volcengine.secret_key=YOUR_SECRET_KEY", + "Alternative (per-shell):", + " export VOLCENGINE_ACCESS_KEY=YOUR_ACCESS_KEY", + " export VOLCENGINE_SECRET_KEY=YOUR_SECRET_KEY", + ] + ) ) + def _get_dotenv_credentials(self, service_key: str) -> Optional[Credentials]: + """Attempt to read credentials from a local .env file. + + This is a last-resort fallback for CLI users who commonly expect `.env` in the + current working directory to provide environment variables. + + Notes: + - Reads only `Path.cwd() / '.env'`. + - Does NOT mutate the current process environment. + """ + + try: + from dotenv import dotenv_values + except Exception: + return None + + env_file_path = Path.cwd() / ".env" + + try: + values = dotenv_values(env_file_path) + except Exception: + return None + + if not isinstance(values, dict): + return None + + def _get(key: str) -> str: + v = values.get(key) + return str(v) if v is not None else "" + + svc_upper = service_key.upper() + + # Service-specific keys (align with environment variable behavior) + ak = _get(f"VOLCENGINE_{svc_upper}_ACCESS_KEY") + sk = _get(f"VOLCENGINE_{svc_upper}_SECRET_KEY") + if not ak or not sk: + # Legacy support + ak = ak or _get(f"VOLC_{svc_upper}_ACCESSKEY") + sk = sk or _get(f"VOLC_{svc_upper}_SECRETKEY") + + if ak and sk: + return Credentials(access_key=ak, secret_key=sk) + + # Global keys + ak = _get("VOLCENGINE_ACCESS_KEY") or _get("VOLC_ACCESSKEY") + sk = _get("VOLCENGINE_SECRET_KEY") or _get("VOLC_SECRETKEY") + + if ak and sk: + return Credentials(access_key=ak, secret_key=sk) + + return None + def _get_service_env_credentials(self, service_key: str) -> Optional[Credentials]: svc_upper = service_key.upper() ak = os.getenv(f"VOLCENGINE_{svc_upper}_ACCESS_KEY") diff --git a/agentkit/toolkit/config/utils.py b/agentkit/toolkit/config/utils.py index 7e17d7f..fab8132 100644 --- a/agentkit/toolkit/config/utils.py +++ b/agentkit/toolkit/config/utils.py @@ -18,7 +18,7 @@ from typing import Dict, Any, Optional import logging -from dotenv import load_dotenv, dotenv_values +from dotenv import dotenv_values from yaml import safe_load, YAMLError from .constants import AUTO_CREATE_VE @@ -48,8 +48,7 @@ def load_dotenv_file(project_dir: Path) -> Dict[str, str]: if not env_file_path.exists(): return {} - # Load .env into environment temporarily to get the values - load_dotenv(env_file_path) + # Parse values without mutating the current process environment. env_values = dotenv_values(env_file_path) return {k: str(v) for k, v in env_values.items() if v is not None} diff --git a/tests/platform/test_configuration_creds.py b/tests/platform/test_configuration_creds.py index d711181..43d18a5 100644 --- a/tests/platform/test_configuration_creds.py +++ b/tests/platform/test_configuration_creds.py @@ -14,6 +14,7 @@ import pytest import os +from pathlib import Path from agentkit.platform.configuration import VolcConfiguration @@ -91,6 +92,101 @@ def test_creds_vefaas_fallback( assert creds.secret_key == "vefaas_sk" assert creds.session_token == "vefaas_token" + def test_creds_dotenv_fallback_from_cwd( + self, clean_env, mock_global_config, monkeypatch, tmp_path, mocker + ): + """Test fallback to .env in current working directory when other sources are missing.""" + # Ensure VeFaaS IAM check fails + mocker.patch("pathlib.Path.exists", return_value=False) + + monkeypatch.chdir(tmp_path) + (tmp_path / ".env").write_text( + "VOLCENGINE_ACCESS_KEY=AK_FROM_DOTENV\nVOLCENGINE_SECRET_KEY=SK_FROM_DOTENV\n", + encoding="utf-8", + ) + + config = VolcConfiguration() + creds = config.get_service_credentials("agentkit") + + assert creds.access_key == "AK_FROM_DOTENV" + assert creds.secret_key == "SK_FROM_DOTENV" + + def test_creds_dotenv_does_not_override_global_env( + self, clean_env, mock_global_config, monkeypatch, tmp_path + ): + """Test that .env fallback never overrides real process environment variables.""" + os.environ["VOLCENGINE_ACCESS_KEY"] = "AK_FROM_ENV" + os.environ["VOLCENGINE_SECRET_KEY"] = "SK_FROM_ENV" + + monkeypatch.chdir(tmp_path) + (tmp_path / ".env").write_text( + "VOLCENGINE_ACCESS_KEY=AK_FROM_DOTENV\nVOLCENGINE_SECRET_KEY=SK_FROM_DOTENV\n", + encoding="utf-8", + ) + + config = VolcConfiguration() + creds = config.get_service_credentials("agentkit") + + assert creds.access_key == "AK_FROM_ENV" + assert creds.secret_key == "SK_FROM_ENV" + + def test_creds_dotenv_does_not_override_global_config( + self, clean_env, mock_global_config, monkeypatch, tmp_path, mocker + ): + """Test that .env fallback never overrides ~/.agentkit/config.yaml credentials.""" + # Ensure VeFaaS IAM check fails + mocker.patch("pathlib.Path.exists", return_value=False) + + mock_global_config.update( + {"volcengine": {"access_key": "AK_FROM_CFG", "secret_key": "SK_FROM_CFG"}} + ) + + monkeypatch.chdir(tmp_path) + (tmp_path / ".env").write_text( + "VOLCENGINE_ACCESS_KEY=AK_FROM_DOTENV\nVOLCENGINE_SECRET_KEY=SK_FROM_DOTENV\n", + encoding="utf-8", + ) + + config = VolcConfiguration() + creds = config.get_service_credentials("agentkit") + + assert creds.access_key == "AK_FROM_CFG" + assert creds.secret_key == "SK_FROM_CFG" + + def test_creds_dotenv_partial_is_ignored( + self, clean_env, mock_global_config, monkeypatch, tmp_path, mocker + ): + """Test that partial .env credentials are ignored and lookup continues.""" + # Ensure VeFaaS IAM check fails + mocker.patch("pathlib.Path.exists", return_value=False) + + monkeypatch.chdir(tmp_path) + (tmp_path / ".env").write_text( + "VOLCENGINE_ACCESS_KEY=AK_ONLY\n", + encoding="utf-8", + ) + + config = VolcConfiguration() + with pytest.raises(ValueError, match="Volcengine credentials not found"): + config.get_service_credentials("agentkit") + + def test_creds_vefaas_takes_priority_over_dotenv( + self, clean_env, mock_global_config, mock_vefaas_file, monkeypatch, tmp_path + ): + """Test that VeFaaS IAM credentials take priority over .env fallback.""" + monkeypatch.chdir(tmp_path) + (tmp_path / ".env").write_text( + "VOLCENGINE_ACCESS_KEY=AK_FROM_DOTENV\nVOLCENGINE_SECRET_KEY=SK_FROM_DOTENV\n", + encoding="utf-8", + ) + + config = VolcConfiguration() + creds = config.get_service_credentials("agentkit") + + assert creds.access_key == "vefaas_ak" + assert creds.secret_key == "vefaas_sk" + assert creds.session_token == "vefaas_token" + def test_creds_missing_error(self, clean_env, mock_global_config, mocker): """Test error raised when no credentials found.""" mocker.patch("pathlib.Path.exists", return_value=False) diff --git a/tests/toolkit/config/test_env_files.py b/tests/toolkit/config/test_env_files.py new file mode 100644 index 0000000..110ebeb --- /dev/null +++ b/tests/toolkit/config/test_env_files.py @@ -0,0 +1,56 @@ +import os + +from agentkit.toolkit.config import CommonConfig +from agentkit.toolkit.config.utils import load_dotenv_file, merge_runtime_envs + + +def test_load_dotenv_file_returns_values(tmp_path, monkeypatch): + env_file = tmp_path / ".env" + env_file.write_text("FOO=bar\nHELLO=world\n", encoding="utf-8") + + monkeypatch.delenv("FOO", raising=False) + monkeypatch.delenv("HELLO", raising=False) + + values = load_dotenv_file(tmp_path) + + assert values["FOO"] == "bar" + assert values["HELLO"] == "world" + + +def test_load_dotenv_file_does_not_mutate_process_environment(tmp_path, monkeypatch): + env_file = tmp_path / ".env" + env_file.write_text( + "FOO=bar\nVOLCENGINE_ACCESS_KEY=ak_from_dotenv\n", + encoding="utf-8", + ) + + monkeypatch.delenv("FOO", raising=False) + monkeypatch.delenv("VOLCENGINE_ACCESS_KEY", raising=False) + + values = load_dotenv_file(tmp_path) + assert values["FOO"] == "bar" + assert values["VOLCENGINE_ACCESS_KEY"] == "ak_from_dotenv" + + assert "FOO" not in os.environ + assert "VOLCENGINE_ACCESS_KEY" not in os.environ + + +def test_merge_runtime_envs_precedence_includes_dotenv(tmp_path): + (tmp_path / "config.yaml").write_text( + "model:\n api_key: from_config\n", + encoding="utf-8", + ) + (tmp_path / ".env").write_text( + "A=dotenv\nB=dotenv\nMODEL_API_KEY=from_env\n", + encoding="utf-8", + ) + + common_config = CommonConfig(runtime_envs={"A": "common", "B": "common"}) + strategy_config = {"runtime_envs": {"B": "strategy", "C": "strategy"}} + + merged = merge_runtime_envs(common_config, strategy_config, project_dir=tmp_path) + + assert merged["A"] == "common" + assert merged["B"] == "strategy" + assert merged["C"] == "strategy" + assert merged["MODEL_API_KEY"] == "from_env"