"""Phase C: Agent Registry API routes.
Provides CRUD operations for registering and managing external
AI agents that can participate in form submission workflows.
"""
from flask import Blueprint, jsonify, request, session
from app.models_agents import (
create_agent,
delete_agent,
get_agent,
list_agent_calls,
list_agents,
update_agent,
)
from app.routes.auth import current_user, login_required
agents_bp = Blueprint("agents", __name__, url_prefix="/api/v2/agents")
# ─── Agent Registry ─────────────────────────────────────────────────────────
@agents_bp.route("", methods=["GET"])
@login_required
def agents_list():
"""GET /api/v2/agents — List all agents for the current user."""
user = current_user()
if not user:
return jsonify({"error": "Not authenticated"}), 401
enabled_only = request.args.get("enabled_only", "false").lower() == "true"
agents = list_agents(user["id"], enabled_only=enabled_only)
# Strip API keys from list response
result = []
for a in agents:
a_copy = dict(a)
a_copy.pop("api_key", None)
result.append(a_copy)
return jsonify({"agents": result})
@agents_bp.route("", methods=["POST"])
@login_required
def agents_create():
"""POST /api/v2/agents — Register a new external agent.
Request body:
name: Agent display name (required)
endpoint_url: Agent HTTP endpoint URL (required)
api_key: Optional API key for authentication
capabilities: List of capability strings (default: [])
metadata: Optional JSON object for custom data
max_timeout_seconds: Max timeout for calls (default: 60)
enabled: Enable/disable agent (default: true)
"""
user = current_user()
if not user:
return jsonify({"error": "Not authenticated"}), 401
data = request.get_json(silent=True) or {}
# Tier gate: check agent quota
from app.models import TIERS, get_user_tier
tier_key, tier_config = get_user_tier(user["id"])
max_agents = tier_config.get("max_agents", 0)
if max_agents == 0:
return jsonify(
{
"error": "Agent registry requires a paid plan. Upgrade to Starter or above.",
"required_tier": "starter",
"current_tier": tier_key,
}
), 402
# Count existing agents
existing_agents = list_agents(user["id"])
if max_agents > 0 and len(existing_agents) >= max_agents:
return jsonify(
{
"error": f"Agent limit reached ({len(existing_agents)}/{max_agents})",
"current_count": len(existing_agents),
"max_agents": max_agents,
"upgrade_available": True,
}
), 402
name = data.get("name", "").strip()
endpoint_url = data.get("endpoint_url", "").strip()
if not name:
return jsonify({"error": "Agent name is required"}), 400
if not endpoint_url:
return jsonify({"error": "Endpoint URL is required"}), 400
# Basic URL validation
if not endpoint_url.startswith(("http://", "https://")):
return jsonify({"error": "Endpoint URL must start with http:// or https://"}), 400
agent_id = create_agent(
user_id=user["id"],
name=name,
endpoint_url=endpoint_url,
api_key=data.get("api_key"),
capabilities=data.get("capabilities", []),
metadata=data.get("metadata"),
max_timeout_seconds=data.get("max_timeout_seconds", 60),
enabled=data.get("enabled", True),
)
if agent_id is None:
return jsonify({"error": "Failed to create agent"}), 500
agent = get_agent(agent_id)
return jsonify(
{
"id": agent_id,
"name": agent["name"],
"endpoint_url": agent["endpoint_url"],
"capabilities": agent["capabilities"],
"metadata": agent["metadata"],
"max_timeout_seconds": agent["max_timeout_seconds"],
"enabled": bool(agent["enabled"]),
"created_at": agent["created_at"],
"api_key_provided": bool(agent.get("api_key")),
}
), 201
@agents_bp.route("/<int:agent_id>", methods=["GET"])
@login_required
def agents_get(agent_id):
"""GET /api/v2/agents/<id> — Get agent details (without API key)."""
user = current_user()
if not user:
return jsonify({"error": "Not authenticated"}), 401
agent = get_agent(agent_id)
if not agent or agent["user_id"] != user["id"]:
return jsonify({"error": "Agent not found"}), 404
result = dict(agent)
result.pop("api_key", None)
return jsonify(result)
@agents_bp.route("/<int:agent_id>", methods=["PUT"])
@login_required
def agents_update(agent_id):
"""PUT /api/v2/agents/<id> — Update agent configuration.
Request body (all optional):
name, endpoint_url, api_key, capabilities,
metadata, max_timeout_seconds, enabled
"""
user = current_user()
if not user:
return jsonify({"error": "Not authenticated"}), 401
# Verify ownership
agent = get_agent(agent_id)
if not agent or agent["user_id"] != user["id"]:
return jsonify({"error": "Agent not found"}), 404
data = request.get_json(silent=True) or {}
valid_fields = {
"name",
"endpoint_url",
"api_key",
"capabilities",
"metadata",
"max_timeout_seconds",
"enabled",
}
updates = {k: v for k, v in data.items() if k in valid_fields and v is not None}
success = update_agent(agent_id, user["id"], **updates)
if not success:
return jsonify({"error": "Failed to update agent"}), 500
updated = get_agent(agent_id)
return jsonify(
{
"id": updated["id"],
"name": updated["name"],
"endpoint_url": updated["endpoint_url"],
"capabilities": updated["capabilities"],
"metadata": updated["metadata"],
"max_timeout_seconds": updated["max_timeout_seconds"],
"enabled": bool(updated["enabled"]),
"updated_at": updated["updated_at"],
}
)
@agents_bp.route("/<int:agent_id>", methods=["DELETE"])
@login_required
def agents_delete(agent_id):
"""DELETE /api/v2/agents/<id> — Remove agent and its call logs."""
user = current_user()
if not user:
return jsonify({"error": "Not authenticated"}), 401
success = delete_agent(agent_id, user["id"])
if not success:
return jsonify({"error": "Agent not found"}), 404
return jsonify({"deleted": True, "id": agent_id})
# ─── Agent Call Logs ────────────────────────────────────────────────────────
@agents_bp.route("/<int:agent_id>/calls", methods=["GET"])
@login_required
def agent_calls_list(agent_id):
"""GET /api/v2/agents/<id>/calls — List call logs for an agent.
Query params:
status: Filter by status (pending, running, completed, failed, timeout)
limit: Max results (default 50)
"""
user = current_user()
if not user:
return jsonify({"error": "Not authenticated"}), 401
# Verify ownership
agent = get_agent(agent_id)
if not agent or agent["user_id"] != user["id"]:
return jsonify({"error": "Agent not found"}), 404
status = request.args.get("status")
limit = min(request.args.get("limit", 50, type=int), 200)
calls = list_agent_calls(user["id"], agent_id=agent_id, limit=limit, status=status)
return jsonify({"calls": calls, "total": len(calls)})
@agents_bp.route("/calls", methods=["GET"])
@login_required
def all_calls_list():
"""GET /api/v2/agents/calls — List all agent call logs for the user.
Query params:
agent_id: Filter by agent
status: Filter by status
limit: Max results (default 50)
"""
user = current_user()
if not user:
return jsonify({"error": "Not authenticated"}), 401
agent_id = request.args.get("agent_id", type=int)
status = request.args.get("status")
limit = min(request.args.get("limit", 50, type=int), 200)
calls = list_agent_calls(user["id"], agent_id=agent_id, limit=limit, status=status)
return jsonify({"calls": calls, "total": len(calls)})