"""Phase C: Agent Registry and Call Log models.
Provides CRUD operations for the agent registry (agents table)
and agent call execution logs (agent_calls table).
"""
import json
import logging
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
logger = logging.getLogger(__name__)
from app.crypto import encrypt_value, try_decrypt
from app.db import _table_exists, get_db
# ─── Column-name allowlist helpers ──────────────────────────────────────
_AGENTS_ALLOWED_COLUMNS = {
"name",
"description",
"model",
"system_prompt",
"instructions",
"config",
"active",
"position",
"icon",
"color",
"temperature",
"max_tokens",
"tools",
"webhook_url",
"api_key",
"endpoint_url",
"capabilities",
"metadata",
"max_timeout_seconds",
"enabled",
"updated_at",
}
_AGENT_CALLS_ALLOWED_COLUMNS = {
"status",
"input",
"output",
"error",
"cost",
"duration",
"model",
"created_at",
"response_payload",
"error_message",
"duration_ms",
"completed_at",
}
def _validate_columns(columns: list, allowed: set) -> list:
"""Validate column names against an allowlist. Returns list of 'col = ?' clauses."""
valid = []
for col in columns:
if col in allowed:
valid.append(f"{col} = ?")
else:
logger.warning(f"Blocked column in UPDATE: {col} (not in allowlist)")
return valid
# ─── Agent Registry ───────────────────────────────────────────────────────────
def create_agent(
user_id: int,
name: str,
endpoint_url: str,
api_key: str | None = None,
capabilities: list[str] | None = None,
metadata: dict[str, Any] | None = None,
max_timeout_seconds: int = 60,
enabled: bool = True,
) -> int | None:
"""Register a new external agent."""
conn = None
try:
conn = get_db()
api_key_encrypted = None
if api_key:
api_key_encrypted = encrypt_value(user_id, api_key)
cursor = conn.execute(
"""
INSERT INTO agents (user_id, name, endpoint_url, api_key_encrypted,
capabilities, metadata, max_timeout_seconds, enabled)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""",
(
user_id,
name,
endpoint_url,
api_key_encrypted,
json.dumps(capabilities or []),
json.dumps(metadata or {}),
max_timeout_seconds,
int(enabled),
),
)
conn.commit()
return cursor.lastrowid
except Exception:
return None
finally:
if conn:
conn.close()
def get_agent(agent_id: int) -> dict[str, Any] | None:
"""Fetch a single agent by ID."""
conn = None
try:
conn = get_db()
row = conn.execute("SELECT * FROM agents WHERE id = ?", (agent_id,)).fetchone()
if not row:
return None
return _row_to_agent(row)
finally:
if conn:
conn.close()
def list_agents(user_id: int, enabled_only: bool = False) -> list[dict[str, Any]]:
"""List all agents for a user."""
conn = None
try:
conn = get_db()
query = "SELECT * FROM agents WHERE user_id = ?"
params: list[Any] = [user_id]
if enabled_only:
query += " AND enabled = 1"
query += " ORDER BY name ASC"
rows = conn.execute(query, params).fetchall()
return [_row_to_agent(row) for row in rows]
finally:
if conn:
conn.close()
def update_agent(agent_id: int, user_id: int, **kwargs: Any) -> bool:
"""Update an agent's configuration."""
conn = None
try:
conn = get_db()
row = conn.execute("SELECT id, user_id FROM agents WHERE id = ?", (agent_id,)).fetchone()
if not row or row["user_id"] != user_id:
return False
updates: list[str] = []
params: list[Any] = []
column_names: list[str] = []
if "name" in kwargs:
column_names.append("name")
params.append(kwargs["name"])
if "endpoint_url" in kwargs:
column_names.append("endpoint_url")
params.append(kwargs["endpoint_url"])
if "api_key" in kwargs:
if kwargs["api_key"]:
column_names.append("api_key_encrypted")
params.append(encrypt_value(user_id, kwargs["api_key"]))
else:
updates.append("api_key_encrypted = NULL")
if "capabilities" in kwargs:
column_names.append("capabilities")
params.append(json.dumps(kwargs["capabilities"] or []))
if "metadata" in kwargs:
column_names.append("metadata")
params.append(json.dumps(kwargs["metadata"] or {}))
if "max_timeout_seconds" in kwargs:
column_names.append("max_timeout_seconds")
params.append(kwargs["max_timeout_seconds"])
if "enabled" in kwargs:
column_names.append("enabled")
params.append(int(kwargs["enabled"]))
# Validate column names against allowlist
validated = _validate_columns(column_names, _AGENTS_ALLOWED_COLUMNS)
updates.extend(validated)
if not updates:
return True
params.append(agent_id)
updates.append("updated_at = CURRENT_TIMESTAMP")
conn.execute(f"UPDATE agents SET {', '.join(updates)} WHERE id = ?", params)
conn.commit()
return True
finally:
if conn:
conn.close()
def delete_agent(agent_id: int, user_id: int) -> bool:
"""Delete an agent from the registry."""
conn = None
try:
conn = get_db()
row = conn.execute("SELECT id, user_id FROM agents WHERE id = ?", (agent_id,)).fetchone()
if not row or row["user_id"] != user_id:
return False
conn.execute("DELETE FROM agent_calls WHERE agent_id = ?", (agent_id,))
conn.execute("DELETE FROM agents WHERE id = ?", (agent_id,))
conn.commit()
return True
finally:
if conn:
conn.close()
def get_agent_with_api_key(agent_id: int, user_id: int) -> dict[str, Any] | None:
"""Fetch agent with decrypted API key for execution context."""
conn = None
try:
conn = get_db()
row = conn.execute(
"SELECT * FROM agents WHERE id = ? AND user_id = ? AND enabled = 1",
(agent_id, user_id),
).fetchone()
if not row:
return None
return _row_to_agent(row)
finally:
if conn:
conn.close()
def get_agent_by_api_key(api_key: str) -> dict[str, Any] | None:
"""Fetch enabled agent by plaintext API key (for callback auth)."""
conn = None
try:
conn = get_db()
row = conn.execute(
"SELECT * FROM agents WHERE api_key_encrypted = ? AND enabled = 1",
(api_key,),
).fetchone()
if not row:
return None
return _row_to_agent(row)
finally:
if conn:
conn.close()
# ─── Agent Call Log ───────────────────────────────────────────────────────────
def log_agent_call(
user_id: int,
agent_id: int,
chain_action_id: int | None = None,
site_id: int | None = None,
submission_id: int | None = None,
request_payload: dict[str, Any] | None = None,
) -> int | None:
"""Create a new agent call log entry."""
conn = None
try:
conn = get_db()
cursor = conn.execute(
"""
INSERT INTO agent_calls (user_id, agent_id, chain_action_id,
site_id, submission_id, request_payload, status)
VALUES (?, ?, ?, ?, ?, ?, 'pending')
""",
(
user_id,
agent_id,
chain_action_id,
site_id,
submission_id,
json.dumps(request_payload) if request_payload else None,
),
)
conn.commit()
return cursor.lastrowid
except Exception:
return None
finally:
if conn:
conn.close()
def update_agent_call(
call_id: int,
status: str,
response_payload: dict[str, Any] | None = None,
error_message: str | None = None,
duration_ms: int | None = None,
) -> bool:
"""Update an agent call log entry."""
conn = None
try:
conn = get_db()
updates: list[str] = ["status = ?"]
params: list[Any] = [status]
if response_payload is not None:
updates.append("response_payload = ?")
params.append(json.dumps(response_payload))
if error_message is not None:
updates.append("error_message = ?")
params.append(str(error_message))
if duration_ms is not None:
updates.append("duration_ms = ?")
params.append(int(duration_ms))
if status in ("completed", "failed", "timeout"):
updates.append("completed_at = CURRENT_TIMESTAMP")
params.append(call_id)
conn.execute(f"UPDATE agent_calls SET {', '.join(updates)} WHERE id = ?", params)
conn.commit()
return True
finally:
if conn:
conn.close()
def get_agent_call(call_id: int) -> dict[str, Any] | None:
"""Fetch a single agent call log entry."""
conn = None
try:
conn = get_db()
row = conn.execute("SELECT * FROM agent_calls WHERE id = ?", (call_id,)).fetchone()
if not row:
return None
return _call_row_to_dict(row)
finally:
if conn:
conn.close()
def list_agent_calls(
user_id: int,
agent_id: int | None = None,
limit: int = 50,
status: str | None = None,
) -> list[dict[str, Any]]:
"""List agent call logs for a user."""
conn = None
try:
conn = get_db()
query = "SELECT * FROM agent_calls WHERE user_id = ?"
params: list[Any] = [user_id]
if agent_id is not None:
query += " AND agent_id = ?"
params.append(agent_id)
if status:
query += " AND status = ?"
params.append(status)
query += " ORDER BY created_at DESC LIMIT ?"
params.append(limit)
rows = conn.execute(query, params).fetchall()
return [_call_row_to_dict(row) for row in rows]
finally:
if conn:
conn.close()
# ─── Helpers ──────────────────────────────────────────────────────────────────
def _row_to_agent(row: dict) -> dict[str, Any]:
"""Convert a database row to an agent dict."""
result = dict(row)
result["capabilities"] = json.loads(row["capabilities"]) if row["capabilities"] else []
result["metadata"] = json.loads(row["metadata"]) if row["metadata"] else {}
if row["api_key_encrypted"]:
decrypted = try_decrypt(row["user_id"], row["api_key_encrypted"])
result["api_key"] = decrypted if decrypted else None
else:
result["api_key"] = None
result.pop("api_key_encrypted", None)
return result
def _call_row_to_dict(row: dict) -> dict[str, Any]:
"""Convert a database row to an agent call dict."""
result = dict(row)
if row["request_payload"]:
try:
result["request_payload"] = json.loads(row["request_payload"])
except (json.JSONDecodeError, TypeError):
result["request_payload"] = row["request_payload"]
else:
result["request_payload"] = None
if row["response_payload"]:
try:
result["response_payload"] = json.loads(row["response_payload"])
except (json.JSONDecodeError, TypeError):
result["response_payload"] = row["response_payload"]
else:
result["response_payload"] = None
return result