agent-control-plane

Safety and approval controls for AI agents.

The control plane decides when/how an agent may act. The data plane executes side effects.

Watch Demo

Watch interactive terminal recording

Why This Exists

Agent stacks handle execution well; governance is the gap. This package provides:

• Deterministic policy enforcement before execution.
• Human/risk approval gates for high-impact actions.
• Budget guardrails and kill-switch semantics.
• Durable event history for audit, replay, and recovery.

Good fit:

• Platform teams running production agent workflows.
• Teams needing explicit human-in-the-loop and policy controls.
• Multi-agent systems requiring auditable decisions.

Less useful:

• One-off demos with no side effects.
• Prompt/tooling projects that do not need governance.

Adoption tiers

Pick the tier that matches your use case — each builds on the previous one.

Tier 1 — Cost tracking only (5 minutes)

Token budget enforcement and per-call usage ledger. No sessions, no governance. Works standalone.

uv run python examples/tenant_budget_tracking.py

Tier 2 — Audit trail (add sessions)

Wrap agent runs in a tracked session. Records cost, duration, and outcome without any approval gates.

uv run python examples/audit_trail.py

Tier 3 — Full governance

Policy enforcement, human approval gates, kill switch, multi-agent revocation.

uv run python examples/quickstart_sync.py

---

Install

pip install agent-control-plane

Local Dev

uv sync --extra dev
make check

Quickstart

Use the runnable sync quickstart:

uv run python examples/quickstart_sync.py

ACP-first continuous loop examples:

uv run python examples/single_agent_continuous_loop.py
uv run python examples/multi_agent_continuous_loop.py
uv run python examples/continuous_loop_governance.py
uv run python examples/long_running_autonomous_agent.py --horizon day

Optional SDK integrations (requires provider SDK + API key):

uv run python examples/openai_agents_sdk_integration.py
uv run python examples/claude_agent_sdk_integration.py

For the narrated terminal walkthrough used in the demo video:

make demo-asciicast-agent

Core Capabilities

• Policy and routing: PolicyEngine, ProposalRouter
• Steering: ActionTier.STEER with corrective SteeringContext guidance and suggested alternatives
• Human approvals: ApprovalGate, scoped ticket decisions
• Budget enforcement: BudgetTracker
• Token governance: TokenBudgetTracker (identity-scoped token/cost budgets), ModelGovernor (model tier access policy)
• Concurrency and kill switches: ConcurrencyGuard, KillSwitch
• Condition trees: ConditionEvaluator with recursive and/or/not rules for policy composition
• Pluggable evaluators: EvaluatorRegistry with entry-point discovery, built-in RegexEvaluator and ListEvaluator
• Parallel evaluation: ParallelPolicyEvaluator with cancel-on-deny semantics
• Egress governance: EgressEvaluator — capability-grant model (destination + capability must both be explicitly allowed)
• Durable events and replay: EventStore
• Session lifecycle and recovery: SessionManager, CrashRecovery, TimeoutEscalation, session state integrity validation
• Host wrappers: ControlPlaneFacade (sync), AsyncControlPlaneFacade (async)

Token Budget Tracking

Identity-scoped, persistent token/cost budgets — useful as a first integration point for consumers (much smaller surface than the full proposal lifecycle).

from agent_control_plane import (
    ControlPlaneSetup, GovernanceConfig, IdentityContext, OrgId, ModelId, TokenUsage,
)

cp = ControlPlaneSetup(database_url, governance=GovernanceConfig(token_budget_configs=[...])).build_async()

async with cp.token_budget_tracker() as tracker:
    identity = IdentityContext(org_id=OrgId("tenant-acme"))
    usage = TokenUsage(
        model_id=ModelId("claude-haiku-4-5"),
        input_tokens=1234,
        output_tokens=567,
        total_tokens=1801,
        estimated_cost_usd=0.0042,  # float accepted, coerced via Decimal(str(...))
    )
    await tracker.record_usage(None, identity, usage)  # session_id is optional

Runnable end-to-end example covering seeding, tenant scoping, exhaustion, and cross-process persistence:

uv run python examples/tenant_budget_tracking.py

Notes:

• session_id=None records usage without a control-plane session FK. Use a real session UUID if you want the TOKEN_USAGE_RECORDED event to land in the event log.
• The context manager opens a fresh DB session and commits on clean exit. For shared-transaction recording, use TokenBudgetTracker.from_session(your_session) instead.
• Pre-call enforcement caveat: tracker.check_budget(...) exists as a pre-call hook, but using it requires you to know the prompt token count in advance. OpenAI-compatible SDKs don't expose this without a separate tokenizer (e.g. tiktoken). For OpenAI-compatible consumers, the practical pattern is post-call record_usage with a local soft-ceiling as defense-in-depth.
• Ledger reflects actual spend, including blocked attempts. record_usage writes the ledger row before raising TokenBudgetExhaustedError, so over-budget calls that already incurred provider cost are still recorded. Callers catching the exception cannot assume "raised ⇒ nothing written" — the row and the TOKEN_USAGE_RECORDED event have already landed. This keeps the ledger accurate as a cost-reporting source under post-call enforcement.

Runtime Notes

• Treat state_bearing=True events as fail-closed.
• Prefer ScopedModelRegistry for production embedding.
• Use SQLite for local/single-process; use Postgres for multi-worker production.

Docs & API

• Architecture: docs/architecture.md
• Operations runbook: docs/operations_runbook.md
• Continuous operation playbook (1h/day/week/month): docs/continuous_operation_playbook.md
• Security model: docs/security_model.md
• Identity integration: docs/integration_identity.md
• Compatibility & upgrading: docs/compatibility.md
• OpenAPI contract (companion gateway): docs/openapi/control-plane-v1.yml
• Public API exports: src/agent_control_plane/init.py

Examples

• Audit trail (Tier 2): examples/audit_trail.py
• Sync quickstart: examples/quickstart_sync.py
• Async quickstart: examples/quickstart.py
• Single-agent continuous loop: examples/single_agent_continuous_loop.py
• Multi-agent continuous loop: examples/multi_agent_continuous_loop.py
• Asciicast sync demo: examples/asciinema_sync_demo.py
• Continuous-loop governance example: examples/continuous_loop_governance.py
• Long-running autonomous example: examples/long_running_autonomous_agent.py
• OpenAI Agents SDK integration: examples/openai_agents_sdk_integration.py
• Claude Agent SDK integration: examples/claude_agent_sdk_integration.py
• Asciicast story runner: scripts/run_asciicast_agent_story.sh
• Audit replay: examples/audit_viewer.py
• Token governance demo: examples/token_governance_demo.py
• MCP gateway demo: examples/mcp_tool_gateway.py
• Steering demo: examples/steering_demo.py
• Condition trees and evaluator plugins: examples/condition_tree_evaluator_demo.py
• Companion REST/dashboard starter: examples/companion_gateway

License

MIT