GitHub - agentcontrol/agent-control: Centralized agent control plane for governing runtime agent behavior at scale. Configurable, extensible, and production-ready.

Start server
  ↓
Install SDK
  ↓
Wrap a model or tool call with @control() and register your agent
  ↓
Create controls (UI or SDK/API)

curl -L https://raw.githubusercontent.com/agentcontrol/agent-control/refs/heads/main/docker-compose.yml | docker compose -f - up -d

export AGENT_CONTROL_SERVER_HOST_PORT=18000
export AGENT_CONTROL_DB_HOST_PORT=15432
export AGENT_CONTROL_API_KEY_ENABLED=true
export AGENT_CONTROL_API_KEYS="agent-api-key"
export AGENT_CONTROL_ADMIN_API_KEYS="admin-api-key"
export AGENT_CONTROL_POSTGRES_PASSWORD="postgres-password"

curl -L https://raw.githubusercontent.com/agentcontrol/agent-control/refs/heads/main/docker-compose.yml | docker compose -f - up -d

curl http://localhost:8000/health

uv venv
source .venv/bin/activate
uv pip install agent-control-sdk

# my_agent.py

import asyncio
import agent_control
from agent_control import control, ControlViolationError

# Protect any function (like LLM calls)

@control()
async def chat(message: str) -> str:
    # In production: response = await LLM.ainvoke(message)
    # For demo: simulate LLM that might leak sensitive data
    if "test" in message.lower():
        return "Your SSN is 123-45-6789"  # Will be blocked!
    return f"Echo: {message}"

# Initialize your agent

agent_control.init(
    agent_name="awesome_bot_3000",  # Unique name
    agent_description="My Chatbot",
)

async def main():
    try:
        print(await chat("test"))  # ❌ Blocked
    except ControlViolationError as e:
        print(f"❌ Blocked: {e.control_name}")
    finally:
        await agent_control.ashutdown()

if __name__ == "__main__":
    asyncio.run(main())

from agent_control import (
    register_control_event_sink,
    unregister_control_event_sink,
)
from agent_control_telemetry import BaseControlEventSink, SinkResult


class MyControlEventSink(BaseControlEventSink):
    def write_events(self, events):
        for event in events:
            forward_to_external_system(event.model_dump(mode="json"))
        return SinkResult(accepted=len(events), dropped=0)


sink = MyControlEventSink()
register_control_event_sink(sink)

# Select registered sinks instead of the default SDK -> server sink.
agent_control.init(
    agent_name="awesome_bot_3000",
    observability_enabled=True,
    observability_sink_name="registered",
)

# Later, when tearing down the integration:
unregister_control_event_sink(sink)

# setup.py - Run once to configure agent controls

import asyncio
from datetime import datetime, UTC
from agent_control import AgentControlClient, controls, agents
from agent_control_models import Agent

async def setup():
    async with AgentControlClient() as client:  # Defaults to localhost:8000
        # 1. Register agent first
        agent = Agent(
            agent_name="awesome_bot_3000",
            agent_description="My Chatbot",
            agent_created_at=datetime.now(UTC).isoformat(),
        )
        await agents.register_agent(client, agent, steps=[])

        # 2. Create control (blocks SSN patterns in output)
        control = await controls.create_control(
            client,
            name="block-ssn",
            data={
                "enabled": True,
                "execution": "server",
                "scope": {"stages": ["post"]},
                "condition": {
                    "selector": {"path": "output"},
                    "evaluator": {
                        "name": "regex",
                        "config": {"pattern": r"\b\d{3}-\d{2}-\d{4}\b"},
                    },
                },
                "action": {"decision": "deny"},
            },
        )

        # 3. Associate control directly with agent
        await agents.add_agent_control(
            client,
            agent_name=agent.agent_name,
            control_id=control["control_id"],
        )

        print("✅ Setup complete!")
        print(f"   Control ID: {control['control_id']}")

asyncio.run(setup())

uv run setup.py
uv run my_agent.py