Agent Controller

A control-plane service for enterprise AI agents: identity, intent-level policy, a single enforcement gateway, task orchestration, and full multi-hop audit — in one FastAPI service you can split apart later.

Why this shape

Agent traffic isn't like normal service traffic: it's dynamic (agents discover tools/agents at runtime), it needs per-agent revocable identity rather than shared service accounts, and policy has to be expressed as intent ("can X write to the CRM") rather than IP:port rules. This controller gives you one place to answer all three.

Components

File	Responsibility
app/identity.py	Agent registry + short-lived JWT issuance (15 min TTL). Swap the bootstrap shared-secret for mTLS/SPIFFE in production.
app/policy.py	Glob-based intent rules (agent → target → action → allow/deny), first match wins, default deny.
app/gateway.py	The single enforcement point. Every call — identity check, policy check, execution, audit log — flows through here.
app/orchestrator.py	Declarative task→route table that drives multi-step agent chains through the Gateway (so orchestrated calls are policy-checked and audited exactly like direct ones).
app/audit.py	In-memory event log keyed by trace_id, so a full multi-hop chain can be reconstructed in call order. Swap the store for Kafka/a DB/SIEM in production.
app/demo_backends.py, app/routes.py	Example tool/agent backends and task routes so the service runs standalone. Replace with real MCP tool servers / agent HTTP endpoints.

Run it

docker compose up --build
# or locally:
pip install -r requirements.txt
AGENT_CONTROLLER_CONFIG=config uvicorn app.main:app --reload

Try it

# 1. Get a short-lived token for an agent
TOKEN=$(curl -s -X POST localhost:8000/auth/token \
  -H "Content-Type: application/json" \
  -d '{"agent_id":"research-agent","shared_secret":"research-agent-dev-secret"}' \
  | python3 -c "import sys,json;print(json.load(sys.stdin)['access_token'])")

# 2. Make an allowed call
curl -s -X POST localhost:8000/invoke -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"target_type":"tool","target_id":"web-search-tool","action":"call","payload":{"query":"edge compute"}}'

# 3. Make a call policy denies (research-agent may never write to CRM)
curl -s -X POST localhost:8000/invoke -H "Authorization: Bearer $TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"target_type":"tool","target_id":"crm-write-tool","action":"call","payload":{"record_id":"123"}}'

# 4. Run a multi-hop orchestrated task (search -> summarize)
ORCH_TOKEN=$(curl -s -X POST localhost:8000/auth/token \
  -H "Content-Type: application/json" \
  -d '{"agent_id":"orchestrator","shared_secret":"orchestrator-dev-secret"}' \
  | python3 -c "import sys,json;print(json.load(sys.stdin)['access_token'])")

curl -s -X POST localhost:8000/tasks -H "Authorization: Bearer $ORCH_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"task_type":"research_and_summarize","payload":{"query":"5G edge compute"},"requested_by":"you"}'

# 5. Reconstruct the full chain from the trace_id returned above
curl -s localhost:8000/audit/trace/<trace_id> -H "Authorization: Bearer $ORCH_TOKEN"

Wiring in real agents/tools

• Tools: replace the functions in demo_backends.py with httpx calls to your MCP tool servers, then gateway.register_backend("your-tool-id", handler).
• Agents: same pattern — an agent-to-agent call is just a backend that POSTs to another agent's endpoint (forward the _trace_id from the payload so the chain stays connected across services).
• Policy: edit config/policies.yaml. Rules are evaluated top-down, first match wins, and anything unmatched is denied by default.
• Identity: edit config/agents.yaml to register new agents. For production, replace the shared-secret bootstrap in identity.py with mTLS client certs or SPIFFE/SPIRE workload identity — the JWT issuance and verification logic downstream doesn't need to change.

Path to production

1. Swap bootstrap secrets for mTLS/SPIFFE workload identity.
2. Swap the in-memory AuditLog for an append-only store (DB, Kafka, or your SIEM) — the interface (record, trace, recent) stays the same.
3. Split Gateway and Orchestrator into separate deployables if load patterns diverge; they only depend on each other through the Gateway's public invoke() method, so this is a clean seam.
4. Add rate limiting and per-agent quotas in the Gateway alongside the policy check.
5. Point demo_backends.py's replacements at real MCP tool servers.

agent-controller