feat: add .env fallback and refine dotenv loading (#50)

Author: innsd

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")

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}
 

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)

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"