516 lines
17 KiB
Python
516 lines
17 KiB
Python
"""Settings API routes with config hot-reload support."""
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
from typing import Any
|
|
|
|
import yaml
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from pydantic import BaseModel
|
|
|
|
from agentkit.server.auth.dependencies import require_permission
|
|
from agentkit.server.auth.permissions import Permission
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(tags=["settings"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper: mask API keys (show only last 4 chars)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _mask_api_key(key: str | None) -> str:
|
|
"""Mask an API key, showing only the last 4 characters."""
|
|
if not key:
|
|
return ""
|
|
if len(key) <= 4:
|
|
return "****"
|
|
return "****" + key[-4:]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pydantic models for request/response
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class LlmProviderResponse(BaseModel):
|
|
name: str
|
|
type: str
|
|
api_key: str # masked
|
|
base_url: str
|
|
models: dict[str, Any]
|
|
max_tokens: int
|
|
timeout: float
|
|
|
|
|
|
class LlmConfigResponse(BaseModel):
|
|
providers: list[LlmProviderResponse]
|
|
model_aliases: dict[str, str]
|
|
fallbacks: dict[str, list[str]]
|
|
|
|
|
|
class LlmProviderUpdate(BaseModel):
|
|
name: str
|
|
type: str = "openai"
|
|
api_key: str | None = None
|
|
base_url: str = ""
|
|
models: dict[str, Any] | None = None
|
|
max_tokens: int = 4096
|
|
timeout: float = 120.0
|
|
|
|
|
|
class LlmConfigUpdate(BaseModel):
|
|
providers: list[LlmProviderUpdate] | None = None
|
|
model_aliases: dict[str, str] | None = None
|
|
fallbacks: dict[str, list[str]] | None = None
|
|
|
|
|
|
class SkillsConfigResponse(BaseModel):
|
|
paths: list[str]
|
|
auto_discover: bool
|
|
|
|
|
|
class SkillsConfigUpdate(BaseModel):
|
|
paths: list[str] | None = None
|
|
auto_discover: bool | None = None
|
|
|
|
|
|
class KbConfigResponse(BaseModel):
|
|
memory: dict[str, Any]
|
|
|
|
|
|
class KbConfigUpdate(BaseModel):
|
|
memory: dict[str, Any]
|
|
|
|
|
|
class GeneralConfigResponse(BaseModel):
|
|
host: str
|
|
port: int
|
|
workers: int
|
|
log_level: str
|
|
log_format: str
|
|
rate_limit: int
|
|
cors_origins: list[str]
|
|
|
|
|
|
class GeneralConfigUpdate(BaseModel):
|
|
host: str | None = None
|
|
port: int | None = None
|
|
workers: int | None = None
|
|
log_level: str | None = None
|
|
log_format: str | None = None
|
|
rate_limit: int | None = None
|
|
cors_origins: list[str] | None = None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper: read/write config file
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _read_yaml_config(config_path: str) -> dict:
|
|
"""Read the YAML config file and return the parsed dict."""
|
|
with open(config_path, encoding="utf-8") as f:
|
|
return yaml.safe_load(f) or {}
|
|
|
|
|
|
def _reverse_resolve_env(data: Any, original: Any) -> Any:
|
|
"""Reverse-resolve env var references: if original had ${VAR} and current
|
|
value matches os.environ[VAR], keep the ${VAR} reference instead of
|
|
writing the plaintext value back to YAML."""
|
|
if isinstance(original, str) and isinstance(data, str):
|
|
# Check if original was an env var reference like ${VAR} or ${VAR:-default}
|
|
env_refs = re.findall(r"\$\{([^}]+)\}", original)
|
|
for expr in env_refs:
|
|
var_name = expr.split(":-")[0] if ":-" in expr else expr
|
|
env_val = os.environ.get(var_name)
|
|
if env_val is not None and data == env_val:
|
|
return original # Keep the ${VAR} reference
|
|
# If original was a ${VAR} reference but value doesn't match env, still keep ref
|
|
if re.match(r"^\$\{[^}]+\}$", original):
|
|
return original
|
|
if isinstance(data, dict) and isinstance(original, dict):
|
|
result = {}
|
|
for k in data:
|
|
result[k] = _reverse_resolve_env(data[k], original.get(k))
|
|
return result
|
|
if isinstance(data, list) and isinstance(original, list):
|
|
# For lists: reverse-resolve matching items, keep new items as-is
|
|
result = []
|
|
for i, item in enumerate(data):
|
|
if i < len(original):
|
|
result.append(_reverse_resolve_env(item, original[i]))
|
|
else:
|
|
result.append(item)
|
|
return result
|
|
# For lists that changed length or type mismatches, return data as-is
|
|
return data
|
|
|
|
|
|
def _write_yaml_config(config_path: str, data: dict) -> None:
|
|
"""Write the full config dict back to the YAML file.
|
|
|
|
Preserves ${VAR} env var references: if the original YAML had ${VAR}
|
|
and the resolved value matches os.environ[VAR], the reference is kept
|
|
instead of writing the plaintext value.
|
|
|
|
Uses ruamel.yaml when available to preserve comments and formatting.
|
|
Falls back to PyYAML otherwise.
|
|
|
|
The write is atomic (write-temp + fsync + os.replace) so a crash
|
|
mid-write leaves the original file intact (U3 — R6).
|
|
"""
|
|
import io
|
|
|
|
from agentkit.server.utils.atomic_write import write_text_atomic
|
|
|
|
original = _read_yaml_config(config_path)
|
|
preserved = _reverse_resolve_env(data, original)
|
|
try:
|
|
from ruamel.yaml import YAML
|
|
|
|
yaml_writer = YAML()
|
|
yaml_writer.default_flow_style = False
|
|
yaml_writer.allow_unicode = True
|
|
# Re-read with ruamel to get the commented structure
|
|
with open(config_path, encoding="utf-8") as f:
|
|
original_data = yaml_writer.load(f)
|
|
# Apply preserved values onto the ruamel-parsed structure
|
|
_deep_update_ruamel(original_data, preserved)
|
|
buf = io.StringIO()
|
|
yaml_writer.dump(original_data, buf)
|
|
write_text_atomic(config_path, buf.getvalue())
|
|
except ImportError:
|
|
buf = io.StringIO()
|
|
yaml.dump(preserved, buf, default_flow_style=False, allow_unicode=True, sort_keys=False)
|
|
write_text_atomic(config_path, buf.getvalue())
|
|
|
|
|
|
def _deep_update_ruamel(target: Any, source: Any) -> None:
|
|
"""Deep-update a ruamel.yaml CommentedMap with values from source dict.
|
|
|
|
Preserves comments and formatting in target while updating values.
|
|
"""
|
|
if not isinstance(source, dict) or not hasattr(target, "__setitem__"):
|
|
return
|
|
for key, value in source.items():
|
|
if key in target and isinstance(value, dict) and hasattr(target[key], "__setitem__"):
|
|
_deep_update_ruamel(target[key], value)
|
|
else:
|
|
target[key] = value
|
|
|
|
|
|
def _get_config_path(request: Request) -> str:
|
|
"""Get the config file path from app state, or raise 400."""
|
|
server_config = request.app.state.server_config
|
|
if server_config is None or not server_config._config_path:
|
|
raise HTTPException(status_code=400, detail="No config file path available")
|
|
return server_config._config_path
|
|
|
|
|
|
def _write_env_var(config_path: str, key: str, value: str) -> None:
|
|
"""Write or update an environment variable in the .env file next to config.
|
|
|
|
If the key already exists in .env, its value is updated in place.
|
|
If not, it's appended. Comments and formatting are preserved.
|
|
|
|
The write is atomic (write-temp + fsync + os.replace) so a crash
|
|
mid-write leaves the original .env intact (U3 — R6).
|
|
"""
|
|
from pathlib import Path
|
|
|
|
from agentkit.server.utils.atomic_write import write_text_atomic
|
|
|
|
env_path = Path(config_path).parent / ".env"
|
|
lines: list[str] = []
|
|
found = False
|
|
|
|
if env_path.exists():
|
|
with open(env_path, encoding="utf-8") as f:
|
|
lines = f.readlines()
|
|
for i, line in enumerate(lines):
|
|
stripped = line.strip()
|
|
if stripped.startswith(f"{key}="):
|
|
lines[i] = f"{key}={value}\n"
|
|
found = True
|
|
break
|
|
else:
|
|
lines = ["# AgentKit Environment Variables\n", "\n"]
|
|
|
|
if not found:
|
|
lines.append(f"{key}={value}\n")
|
|
|
|
write_text_atomic(env_path, "".join(lines))
|
|
|
|
# Also set in current process so the next from_yaml resolves correctly
|
|
os.environ[key] = value
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LLM Settings
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/settings/llm", response_model=LlmConfigResponse)
|
|
async def get_llm_settings(request: Request):
|
|
"""Return LLM config with masked API keys."""
|
|
config = request.app.state.server_config
|
|
if config is None:
|
|
raise HTTPException(status_code=400, detail="Server config not available")
|
|
|
|
llm_config = config.llm_config
|
|
providers = []
|
|
for name, pconf in llm_config.providers.items():
|
|
providers.append(
|
|
LlmProviderResponse(
|
|
name=name,
|
|
type=pconf.type,
|
|
api_key=_mask_api_key(pconf.api_key),
|
|
base_url=pconf.base_url or "",
|
|
models=pconf.models,
|
|
max_tokens=pconf.max_tokens,
|
|
timeout=pconf.timeout,
|
|
)
|
|
)
|
|
|
|
return LlmConfigResponse(
|
|
providers=providers,
|
|
model_aliases=llm_config.model_aliases,
|
|
fallbacks=llm_config.fallbacks,
|
|
)
|
|
|
|
|
|
@router.put("/settings/llm", response_model=LlmConfigResponse)
|
|
async def update_llm_settings(
|
|
request: Request,
|
|
update: LlmConfigUpdate,
|
|
_user=Depends(require_permission(Permission.SYSTEM_CONFIG)),
|
|
):
|
|
"""Update LLM config and trigger hot reload via config file write.
|
|
|
|
When a new plaintext API key is provided, it is stored in .env (not
|
|
agentkit.yaml) and the yaml reference is set to ${ENV_KEY}.
|
|
"""
|
|
config_path = _get_config_path(request)
|
|
data = _read_yaml_config(config_path)
|
|
|
|
# Ensure llm section exists
|
|
if "llm" not in data:
|
|
data["llm"] = {}
|
|
|
|
if update.providers is not None:
|
|
providers_data = {}
|
|
for p in update.providers:
|
|
p_dict: dict[str, Any] = {
|
|
"type": p.type,
|
|
"base_url": p.base_url,
|
|
"max_tokens": p.max_tokens,
|
|
"timeout": p.timeout,
|
|
}
|
|
if p.api_key is not None:
|
|
if p.api_key.startswith("****"):
|
|
# Masked key — keep existing yaml value (preserves ${VAR} refs)
|
|
existing = data.get("llm", {}).get("providers", {}).get(p.name, {})
|
|
p_dict["api_key"] = existing.get("api_key", "")
|
|
else:
|
|
# New plaintext key — write to .env, keep ${VAR} ref in yaml
|
|
# Extract env_key from existing ${VAR} reference if present
|
|
existing = data.get("llm", {}).get("providers", {}).get(p.name, {})
|
|
existing_key = existing.get("api_key", "")
|
|
env_match = re.match(r"^\$\{([^}]+)\}$", str(existing_key))
|
|
if env_match:
|
|
env_key = env_match.group(1).split(":-")[0]
|
|
else:
|
|
env_key = f"{p.name.upper().replace('-', '_')}_API_KEY"
|
|
_write_env_var(config_path, env_key, p.api_key)
|
|
p_dict["api_key"] = f"${{{env_key}}}"
|
|
else:
|
|
# Keep existing key if not provided
|
|
existing = data.get("llm", {}).get("providers", {}).get(p.name, {})
|
|
p_dict["api_key"] = existing.get("api_key", "")
|
|
if p.models is not None:
|
|
p_dict["models"] = p.models
|
|
providers_data[p.name] = p_dict
|
|
data["llm"]["providers"] = providers_data
|
|
|
|
if update.model_aliases is not None:
|
|
data["llm"]["model_aliases"] = update.model_aliases
|
|
|
|
if update.fallbacks is not None:
|
|
data["llm"]["fallbacks"] = update.fallbacks
|
|
|
|
_write_yaml_config(config_path, data)
|
|
# watchfiles will detect the change and trigger _on_config_change
|
|
|
|
# Return updated config (re-read from server_config which may not have
|
|
# reloaded yet, so we return what we just wrote)
|
|
config = request.app.state.server_config
|
|
llm_config = config.llm_config
|
|
providers = []
|
|
for name, pconf in llm_config.providers.items():
|
|
providers.append(
|
|
LlmProviderResponse(
|
|
name=name,
|
|
type=pconf.type,
|
|
api_key=_mask_api_key(pconf.api_key),
|
|
base_url=pconf.base_url or "",
|
|
models=pconf.models,
|
|
max_tokens=pconf.max_tokens,
|
|
timeout=pconf.timeout,
|
|
)
|
|
)
|
|
|
|
return LlmConfigResponse(
|
|
providers=providers,
|
|
model_aliases=llm_config.model_aliases,
|
|
fallbacks=llm_config.fallbacks,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Skills Settings
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/settings/skills", response_model=SkillsConfigResponse)
|
|
async def get_skills_settings(request: Request):
|
|
"""Return skill paths config."""
|
|
config = request.app.state.server_config
|
|
if config is None:
|
|
raise HTTPException(status_code=400, detail="Server config not available")
|
|
|
|
return SkillsConfigResponse(
|
|
paths=config.skill_paths or [],
|
|
auto_discover=config.auto_discover_skills,
|
|
)
|
|
|
|
|
|
@router.put("/settings/skills", response_model=SkillsConfigResponse)
|
|
async def update_skills_settings(
|
|
request: Request,
|
|
update: SkillsConfigUpdate,
|
|
_user=Depends(require_permission(Permission.SYSTEM_CONFIG)),
|
|
):
|
|
"""Update skill paths and trigger hot reload."""
|
|
config_path = _get_config_path(request)
|
|
data = _read_yaml_config(config_path)
|
|
|
|
if "skills" not in data:
|
|
data["skills"] = {}
|
|
|
|
if update.paths is not None:
|
|
data["skills"]["paths"] = update.paths
|
|
if update.auto_discover is not None:
|
|
data["skills"]["auto_discover"] = update.auto_discover
|
|
|
|
_write_yaml_config(config_path, data)
|
|
|
|
config = request.app.state.server_config
|
|
return SkillsConfigResponse(
|
|
paths=config.skill_paths or [],
|
|
auto_discover=config.auto_discover_skills,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Knowledge Base Settings
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/settings/kb", response_model=KbConfigResponse)
|
|
async def get_kb_settings(request: Request):
|
|
"""Return knowledge base connection config."""
|
|
config = request.app.state.server_config
|
|
if config is None:
|
|
raise HTTPException(status_code=400, detail="Server config not available")
|
|
|
|
return KbConfigResponse(memory=config.memory or {})
|
|
|
|
|
|
@router.put("/settings/kb", response_model=KbConfigResponse)
|
|
async def update_kb_settings(
|
|
request: Request,
|
|
update: KbConfigUpdate,
|
|
_user=Depends(require_permission(Permission.SYSTEM_CONFIG)),
|
|
):
|
|
"""Update KB config and trigger hot reload."""
|
|
config_path = _get_config_path(request)
|
|
data = _read_yaml_config(config_path)
|
|
|
|
data["memory"] = update.memory
|
|
|
|
_write_yaml_config(config_path, data)
|
|
|
|
config = request.app.state.server_config
|
|
return KbConfigResponse(memory=config.memory or {})
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# General Settings
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@router.get("/settings/general", response_model=GeneralConfigResponse)
|
|
async def get_general_settings(request: Request):
|
|
"""Return general settings (log level, server port, etc.)."""
|
|
config = request.app.state.server_config
|
|
if config is None:
|
|
raise HTTPException(status_code=400, detail="Server config not available")
|
|
|
|
return GeneralConfigResponse(
|
|
host=config.host,
|
|
port=config.port,
|
|
workers=config.workers,
|
|
log_level=config.log_level,
|
|
log_format=config.log_format,
|
|
rate_limit=config.rate_limit,
|
|
cors_origins=config.cors_origins or ["*"],
|
|
)
|
|
|
|
|
|
@router.put("/settings/general", response_model=GeneralConfigResponse)
|
|
async def update_general_settings(
|
|
request: Request,
|
|
update: GeneralConfigUpdate,
|
|
_user=Depends(require_permission(Permission.SYSTEM_CONFIG)),
|
|
):
|
|
"""Update general settings and trigger hot reload."""
|
|
config_path = _get_config_path(request)
|
|
data = _read_yaml_config(config_path)
|
|
|
|
if "server" not in data:
|
|
data["server"] = {}
|
|
if "logging" not in data:
|
|
data["logging"] = {}
|
|
|
|
if update.host is not None:
|
|
data["server"]["host"] = update.host
|
|
if update.port is not None:
|
|
data["server"]["port"] = update.port
|
|
if update.workers is not None:
|
|
data["server"]["workers"] = update.workers
|
|
if update.rate_limit is not None:
|
|
data["server"]["rate_limit"] = update.rate_limit
|
|
if update.cors_origins is not None:
|
|
data["server"]["cors_origins"] = update.cors_origins
|
|
if update.log_level is not None:
|
|
data["logging"]["level"] = update.log_level
|
|
if update.log_format is not None:
|
|
data["logging"]["format"] = update.log_format
|
|
|
|
_write_yaml_config(config_path, data)
|
|
|
|
config = request.app.state.server_config
|
|
return GeneralConfigResponse(
|
|
host=config.host,
|
|
port=config.port,
|
|
workers=config.workers,
|
|
log_level=config.log_level,
|
|
log_format=config.log_format,
|
|
rate_limit=config.rate_limit,
|
|
cors_origins=config.cors_origins or ["*"],
|
|
)
|