117 lines
4.4 KiB
Python
117 lines
4.4 KiB
Python
"""Environment-backed configuration for the Open Agent API integration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Dict, Mapping, Optional
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
DEFAULT_BASE_URL = "https://superagent.nianxx.cn"
|
|
DEFAULT_SESSION_DB = "runtime/agent_sessions.sqlite3"
|
|
SUPPORTED_AUTH_MODES = {"bearer", "x-api-key"}
|
|
PLACEHOLDER_API_KEYS = {"changeme", "df_open_replace_me", "df_open_xxx"}
|
|
LOCAL_HTTP_HOSTS = {"localhost", "127.0.0.1", "::1"}
|
|
|
|
|
|
class AgentConfigurationError(ValueError):
|
|
"""Raised when required Agent integration configuration is invalid."""
|
|
|
|
|
|
def _positive_float(name: str, raw_value: str) -> float:
|
|
try:
|
|
value = float(raw_value)
|
|
except (TypeError, ValueError) as exc:
|
|
raise AgentConfigurationError(f"{name} must be a number") from exc
|
|
if value <= 0:
|
|
raise AgentConfigurationError(f"{name} must be greater than zero")
|
|
return value
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class AgentConfig:
|
|
"""Validated runtime configuration with no implicit dotenv loading."""
|
|
|
|
base_url: str = DEFAULT_BASE_URL
|
|
api_key: Optional[str] = None
|
|
auth_mode: str = "bearer"
|
|
timeout_seconds: float = 60.0
|
|
connect_timeout_seconds: float = 10.0
|
|
session_db: Path = Path(DEFAULT_SESSION_DB)
|
|
|
|
def __post_init__(self) -> None:
|
|
base_url = self.base_url.strip().rstrip("/")
|
|
parsed = urlparse(base_url)
|
|
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
|
raise AgentConfigurationError("DEERFLOW_BASE_URL must be an absolute http(s) URL")
|
|
if parsed.scheme == "http" and parsed.hostname not in LOCAL_HTTP_HOSTS:
|
|
raise AgentConfigurationError(
|
|
"DEERFLOW_BASE_URL must use HTTPS unless it points to localhost"
|
|
)
|
|
|
|
auth_mode = self.auth_mode.strip().lower()
|
|
if auth_mode not in SUPPORTED_AUTH_MODES:
|
|
raise AgentConfigurationError("DEERFLOW_AUTH_MODE must be 'bearer' or 'x-api-key'")
|
|
|
|
api_key = self.api_key.strip() if self.api_key else None
|
|
if self.timeout_seconds <= 0:
|
|
raise AgentConfigurationError("DEERFLOW_TIMEOUT_SECONDS must be greater than zero")
|
|
if self.connect_timeout_seconds <= 0:
|
|
raise AgentConfigurationError("DEERFLOW_CONNECT_TIMEOUT_SECONDS must be greater than zero")
|
|
|
|
object.__setattr__(self, "base_url", base_url)
|
|
object.__setattr__(self, "auth_mode", auth_mode)
|
|
object.__setattr__(self, "api_key", api_key)
|
|
object.__setattr__(self, "session_db", Path(self.session_db))
|
|
|
|
@classmethod
|
|
def from_env(
|
|
cls,
|
|
env: Optional[Mapping[str, str]] = None,
|
|
*,
|
|
require_api_key: bool = True,
|
|
) -> "AgentConfig":
|
|
source = os.environ if env is None else env
|
|
api_key = source.get("DEERFLOW_OPEN_API_KEY", "").strip() or None
|
|
if require_api_key and not cls._is_real_api_key(api_key):
|
|
raise AgentConfigurationError(
|
|
"DEERFLOW_OPEN_API_KEY is required and must replace the example placeholder"
|
|
)
|
|
|
|
return cls(
|
|
base_url=source.get("DEERFLOW_BASE_URL", DEFAULT_BASE_URL),
|
|
api_key=api_key,
|
|
auth_mode=source.get("DEERFLOW_AUTH_MODE", "bearer"),
|
|
timeout_seconds=_positive_float(
|
|
"DEERFLOW_TIMEOUT_SECONDS",
|
|
source.get("DEERFLOW_TIMEOUT_SECONDS", "60"),
|
|
),
|
|
connect_timeout_seconds=_positive_float(
|
|
"DEERFLOW_CONNECT_TIMEOUT_SECONDS",
|
|
source.get("DEERFLOW_CONNECT_TIMEOUT_SECONDS", "10"),
|
|
),
|
|
session_db=Path(source.get("DEERFLOW_SESSION_DB", DEFAULT_SESSION_DB)),
|
|
)
|
|
|
|
def redacted_summary(self) -> Dict[str, Any]:
|
|
"""Return configuration diagnostics without exposing the credential."""
|
|
|
|
return {
|
|
"base_url": self.base_url,
|
|
"auth_mode": self.auth_mode,
|
|
"api_key_configured": self.api_key_configured,
|
|
"timeout_seconds": self.timeout_seconds,
|
|
"connect_timeout_seconds": self.connect_timeout_seconds,
|
|
"session_db": str(self.session_db),
|
|
}
|
|
|
|
@property
|
|
def api_key_configured(self) -> bool:
|
|
return self._is_real_api_key(self.api_key)
|
|
|
|
@staticmethod
|
|
def _is_real_api_key(api_key: Optional[str]) -> bool:
|
|
return bool(api_key and api_key.strip().lower() not in PLACEHOLDER_API_KEYS)
|