"""Phase C: Agent Protocol.
HTTP endpoints for external agents to interact with AgentForms.
Defines the callback contract for async agent operations and
provides the protocol specification for agent developers.
"""
import hashlib
import hmac
import json
import logging
from datetime import UTC, datetime, timezone
from flask import Blueprint, jsonify, request
from app.models_agents import (
get_agent_by_api_key,
update_agent_call,
)
logger = logging.getLogger("agentforms.protocol")
agent_protocol_bp = Blueprint("agent_protocol", __name__, url_prefix="/api/v2/agent-protocol")
# ─── HMAC signature verification ─────────────────────────────────────────────
def _verify_signature(payload_bytes: bytes, signature: str, secret: str) -> bool:
"""Verify HMAC-SHA256 signature on incoming payload."""
expected = hmac.new(secret.encode("utf-8"), payload_bytes, hashlib.sha256).hexdigest()
return hmac.compare_digest(expected, signature)
def _get_secret(agent: dict) -> str:
"""Extract webhook secret for signature verification."""
metadata = agent.get("metadata") or {}
if isinstance(metadata, str):
try:
metadata = json.loads(metadata)
except json.JSONDecodeError:
metadata = {}
return metadata.get("webhook_secret", "")
# ─── Agent callback endpoint ─────────────────────────────────────────────────
@agent_protocol_bp.route("/callback", methods=["POST"])
def agent_callback():
"""POST /api/v2/agent-protocol/callback — Async agent callback.
External agents use this to report results back when operating
asynchronously (i.e., they returned 202 from the initial call and
need time to process before reporting results).
Headers:
Authorization: Bearer <api_key>
X-Signature-Sha256: <hmac> (optional, if webhook_secret set)
Content-Type: application/json
Body:
call_id: Agent call log ID from the original request
status: "completed" | "failed"
output: Result payload (dict)
error: Error message (on failure)
duration_ms: Processing time in milliseconds
"""
auth_header = request.headers.get("Authorization", "")
if not auth_header.startswith("Bearer "):
return jsonify({"error": "Missing Bearer token"}), 401
api_key = auth_header[7:]
# Look up agent by API key
agent = get_agent_by_api_key(api_key)
if not agent:
return jsonify({"error": "Invalid API key"}), 403
# Verify signature if secret is configured
secret = _get_secret(agent)
if secret:
sig = request.headers.get("X-Signature-Sha256", "")
if not sig or not _verify_signature(request.data, sig, secret):
return jsonify({"error": "Invalid signature"}), 401
data = request.get_json(silent=True) or {}
call_id = data.get("call_id")
if not call_id:
return jsonify({"error": "Missing call_id"}), 400
# Update the call log
try:
update_agent_call(
call_id=int(call_id),
status=data.get("status", "completed"),
response_payload=data.get("output"),
error_message=data.get("error"),
duration_ms=data.get("duration_ms"),
)
return jsonify({"status": "received", "call_id": call_id})
except Exception as e:
logger.error("Callback processing error: %s", e)
return jsonify({"error": "Failed to process callback"}), 500
# ─── Agent discovery / protocol spec ─────────────────────────────────────────
@agent_protocol_bp.route("/spec", methods=["GET"])
def protocol_spec():
"""GET /api/v2/agent-protocol/spec — Agent protocol specification.
Returns the protocol contract that external agents must follow.
Public endpoint — no auth required.
"""
return jsonify(
{
"version": "1.0",
"name": "AgentForms Agent Protocol",
"description": "Protocol for external AI agents to participate in AgentForms workflows.",
"request_format": {
"method": "POST",
"content_type": "application/json",
"headers": {
"Authorization": "Bearer <api_key>",
},
"body": {
"type": "agent_call",
"site_id": "string — form site ID",
"submission_id": "string — submission ID",
"submission_data": {
"customer_name": "string",
"customer_email": "string",
"customer_phone": "string",
"data": "string — JSON-encoded form field data",
"submitted_at": "ISO 8601 timestamp",
},
"chain_context": "object — accumulated context from previous steps",
"prompt": "string — instructions from the chain step config",
},
},
"response_format": {
"synchronous": {
"status_code": 200,
"body": {
"output": "object — structured result",
"error": "string — error message on failure (optional)",
},
},
"asynchronous": {
"initial_response": {
"status_code": 202,
"body": {
"status": "accepted",
"call_id": "number — callback reference ID",
},
},
"callback_endpoint": "/api/v2/agent-protocol/callback",
"callback_body": {
"call_id": "number",
"status": '"completed" | "failed"',
"output": "object — result payload",
"error": "string — on failure",
"duration_ms": "number",
},
},
},
"security": {
"api_key": "Bearer token in Authorization header",
"webhook_secret": "Optional HMAC-SHA256 via X-Signature-Sha256",
},
"timeouts": {
"default_max_seconds": 60,
"configurable": True,
},
}
)
# ─── Health check for agents ─────────────────────────────────────────────────
@agent_protocol_bp.route("/health", methods=["GET"])
def protocol_health():
"""GET /api/v2/agent-protocol/health — Protocol service health."""
return jsonify(
{
"status": "healthy",
"version": "1.0",
"timestamp": datetime.now(UTC).isoformat(),
}
)