simple-agent/app/config.py

import os
from dataclasses import dataclass

from dotenv import load_dotenv

load_dotenv()

GOOGLE_SCOPES = (
    "https://www.googleapis.com/auth/gmail.modify",
    "https://www.googleapis.com/auth/gmail.send",
    "https://www.googleapis.com/auth/calendar.readonly",
)


@dataclass(frozen=True)
class Settings:
    google_client_secrets_file: str
    google_token_file: str
    gmail_scan_interval_minutes: int
    gmail_query: str
    agent_api_key: str
    auth_mode: str
    auth_jwt_secret: str
    auth_jwt_issuer: str | None
    auth_jwt_audience: str | None
    strands_api_key: str
    strands_model_id: str
    strands_base_url: str | None
    strands_timeout_seconds: float
    strands_temperature: float
    llm_fallback_to_rules: bool
    unsubscribe_digest_interval_minutes: int
    unsubscribe_query: str
    unsubscribe_max_results: int
    unsubscribe_state_file: str
    unsubscribe_digest_recipient: str | None
    unsubscribe_send_empty_digest: bool
    unsubscribe_hil_query: str
    unsubscribe_hil_max_results: int
    unsubscribe_hil_state_file: str
    unsubscribe_auto_enabled: bool
    unsubscribe_auto_interval_minutes: int
    unsubscribe_http_timeout_seconds: float
    unsubscribe_user_agent: str
    a2a_public_base_url: str | None
    a2a_agent_name: str
    a2a_agent_description: str
    mcp_auth_mode: str
    mcp_oauth_introspection_url: str | None
    mcp_oauth_client_id: str | None
    mcp_oauth_client_secret: str
    mcp_oauth_issuer: str | None
    mcp_oauth_audience: str | None
    mcp_oauth_timeout_seconds: float
    mcp_enable_mutation_tools: bool
    log_level: str


def get_settings() -> Settings:
    strands_base_url = _first_set_env("STRANDS_OPENAI_BASE_URL", "LLM_BASE_URL").strip()
    unsubscribe_digest_recipient = os.getenv("UNSUBSCRIBE_DIGEST_RECIPIENT", "").strip()
    return Settings(
        google_client_secrets_file=os.getenv("GOOGLE_CLIENT_SECRETS_FILE", "credentials.json"),
        google_token_file=os.getenv("GOOGLE_TOKEN_FILE", "token.json"),
        gmail_scan_interval_minutes=int(os.getenv("GMAIL_SCAN_INTERVAL_MINUTES", "5")),
        gmail_query=os.getenv(
            "GMAIL_QUERY", "in:inbox is:unread -label:AgentProcessed"
        ),
        agent_api_key=os.getenv("AGENT_API_KEY", ""),
        auth_mode=_normalize_auth_mode(os.getenv("AUTH_MODE", "api_key")),
        auth_jwt_secret=os.getenv("AUTH_JWT_SECRET", "").strip(),
        auth_jwt_issuer=os.getenv("AUTH_JWT_ISSUER", "").strip() or None,
        auth_jwt_audience=os.getenv("AUTH_JWT_AUDIENCE", "").strip() or None,
        strands_api_key=_first_set_env("STRANDS_OPENAI_API_KEY", "LLM_API_KEY"),
        strands_model_id=_first_set_env("STRANDS_MODEL_ID", "LLM_MODEL") or "gpt-4.1-mini",
        strands_base_url=strands_base_url or None,
        strands_timeout_seconds=float(
            _first_set_env("STRANDS_TIMEOUT_SECONDS", "LLM_TIMEOUT_SECONDS") or "20"
        ),
        strands_temperature=float(os.getenv("STRANDS_TEMPERATURE", "0")),
        llm_fallback_to_rules=_as_bool(os.getenv("LLM_FALLBACK_TO_RULES", "false")),
        unsubscribe_digest_interval_minutes=int(
            os.getenv("UNSUBSCRIBE_DIGEST_INTERVAL_MINUTES", "1440")
        ),
        unsubscribe_query=os.getenv("UNSUBSCRIBE_QUERY", "label:Advertising"),
        unsubscribe_max_results=int(os.getenv("UNSUBSCRIBE_MAX_RESULTS", "500")),
        unsubscribe_state_file=os.getenv(
            "UNSUBSCRIBE_STATE_FILE", "data/sent_unsubscribe_links.json"
        ),
        unsubscribe_digest_recipient=unsubscribe_digest_recipient or None,
        unsubscribe_send_empty_digest=_as_bool(
            os.getenv("UNSUBSCRIBE_SEND_EMPTY_DIGEST", "false")
        ),
        unsubscribe_hil_query=os.getenv("UNSUBSCRIBE_HIL_QUERY", "label:Advertising"),
        unsubscribe_hil_max_results=int(os.getenv("UNSUBSCRIBE_HIL_MAX_RESULTS", "500")),
        unsubscribe_hil_state_file=os.getenv(
            "UNSUBSCRIBE_HIL_STATE_FILE", "data/unsubscribed_methods.json"
        ),
        unsubscribe_auto_enabled=_as_bool(os.getenv("UNSUBSCRIBE_AUTO_ENABLED", "true")),
        unsubscribe_auto_interval_minutes=int(
            os.getenv("UNSUBSCRIBE_AUTO_INTERVAL_MINUTES", "720")
        ),
        unsubscribe_http_timeout_seconds=float(
            os.getenv("UNSUBSCRIBE_HTTP_TIMEOUT_SECONDS", "12")
        ),
        unsubscribe_user_agent=os.getenv(
            "UNSUBSCRIBE_USER_AGENT",
            "Mozilla/5.0 (compatible; PersonalAgentUnsubscribe/1.0; +https://example.local)",
        ),
        a2a_public_base_url=os.getenv("A2A_PUBLIC_BASE_URL", "").strip() or None,
        a2a_agent_name=os.getenv("A2A_AGENT_NAME", "Personal Agent"),
        a2a_agent_description=os.getenv(
            "A2A_AGENT_DESCRIPTION",
            "Personal productivity agent for calendar availability and email operations.",
        ),
        mcp_auth_mode=_normalize_mcp_auth_mode(os.getenv("MCP_AUTH_MODE", "inherit")),
        mcp_oauth_introspection_url=os.getenv("MCP_OAUTH_INTROSPECTION_URL", "").strip() or None,
        mcp_oauth_client_id=os.getenv("MCP_OAUTH_CLIENT_ID", "").strip() or None,
        mcp_oauth_client_secret=os.getenv("MCP_OAUTH_CLIENT_SECRET", "").strip(),
        mcp_oauth_issuer=os.getenv("MCP_OAUTH_ISSUER", "").strip() or None,
        mcp_oauth_audience=os.getenv("MCP_OAUTH_AUDIENCE", "").strip() or None,
        mcp_oauth_timeout_seconds=float(os.getenv("MCP_OAUTH_TIMEOUT_SECONDS", "8")),
        mcp_enable_mutation_tools=_as_bool(os.getenv("MCP_ENABLE_MUTATION_TOOLS", "false")),
        log_level=os.getenv("LOG_LEVEL", "INFO"),
    )


def _as_bool(value: str) -> bool:
    return value.strip().lower() in {"1", "true", "yes", "on"}


def _first_set_env(*names: str) -> str:
    for name in names:
        value = os.getenv(name)
        if value:
            return value.strip()
    return ""


def _normalize_auth_mode(value: str) -> str:
    normalized = value.strip().lower()
    if normalized in {"api_key", "jwt", "hybrid"}:
        return normalized
    return "api_key"


def _normalize_mcp_auth_mode(value: str) -> str:
    normalized = value.strip().lower()
    if normalized in {"inherit", "api_key", "jwt", "hybrid", "oauth"}:
        return normalized
    return "inherit"