Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
70 changes: 68 additions & 2 deletions agentkit/platform/configuration.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand All @@ -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")
Expand Down
5 changes: 2 additions & 3 deletions agentkit/toolkit/config/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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}

Expand Down
96 changes: 96 additions & 0 deletions tests/platform/test_configuration_creds.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@

import pytest
import os
from pathlib import Path
from agentkit.platform.configuration import VolcConfiguration


Expand Down Expand Up @@ -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)
Expand Down
56 changes: 56 additions & 0 deletions tests/toolkit/config/test_env_files.py
Original file line number Diff line number Diff line change
@@ -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"