Agent Control Panel SDK for Python

Connect an AI agent that already runs in your app to Agent Control Panel (ACP): report every run (tokens, model, latency, errors, the answer) and let your team pause, resume, stop, restart, swap the model or push an approved prompt without redeploying.

• Fail-open. If ACP is slow or down, your agent keeps working. Reporting is buffered and retried on a background thread; nothing the SDK does raises into your code (except AgentPausedError, on purpose).
• Signed control. Commands are HMAC-SHA256 signed with a timestamp and a single-use nonce; unsigned, replayed, stale or wrong-agent commands are rejected.
• Durable pause and model swaps. A pause or a model swap set in ACP survives restarts, extra workers and missed webhooks: every heartbeat reply carries the state ACP holds, and the client adopts it.
• Zero runtime dependencies. Python 3.10+, standard library only, fully typed (py.typed).

Same behaviour and wire contract as the Node SDK (agent-control-panel on npm); both are tested against the same contract vectors.

pip install agent-control-panel

Quickstart

import os
from agent_control_panel import AgentPausedError, create_client

acp = create_client(
    url=os.environ["ACP_URL"],                      # your control panel URL
    api_key=os.environ["ACP_API_KEY"],              # shown once when the key is created
    agent_id=os.environ["ACP_AGENT_ID"],            # from the app page in ACP
    webhook_secret=os.environ["ACP_WEBHOOK_SECRET"],
)
acp.start()  # heartbeats every 30 s, flushes every 5 s, on daemon threads

def handle(question: str) -> str:
    try:
        return acp.wrap_agent_run(
            lambda: my_agent(question, system_prompt=acp.fetch_prompt() or DEFAULT_PROMPT),
            user_input=question,
            model_used=acp.get_effective_model("gpt-4o-mini"),
        )
    except AgentPausedError:
        return "This assistant is paused. Please try again later."

wrap_agent_run records duration, success or error, token usage and the answer text, read from the response objects of the OpenAI, Anthropic and Google Gemini SDKs (and plain dicts of the same shape). For anything else, return a string or call report_run(...) yourself. While the agent is paused or stopped it raises AgentPausedError without calling your function, so a pause really stops model calls.

ACP's quality reviews, Test Lab and error explanations read the user input, the full input and the output, so pass user_input and full_input for accurate results (each field is kept up to 100,000 characters). To send metadata only, create the client with capture_content=False.

OpenAI

from openai import OpenAI

openai = OpenAI()

def answer(question: str) -> str:
    model = acp.get_effective_model("gpt-4o-mini")
    messages = [
        {"role": "system", "content": acp.fetch_prompt() or "You are a helpful assistant."},
        {"role": "user", "content": question},
    ]
    completion = acp.wrap_agent_run(
        lambda: openai.chat.completions.create(model=model, messages=messages),
        user_input=question,
        full_input=messages,
        model_used=model,
    )
    return completion.choices[0].message.content or ""

Anthropic (async)

import asyncio
from anthropic import AsyncAnthropic

claude = AsyncAnthropic()

async def answer(question: str) -> str:
    model = acp.get_effective_model("claude-sonnet-4-5")
    system = await asyncio.to_thread(acp.fetch_prompt)  # fetch_prompt blocks up to 5 s
    messages = [{"role": "user", "content": question}]
    message = await acp.wrap_agent_run_async(
        lambda: claude.messages.create(model=model, max_tokens=1024, system=system, messages=messages),
        user_input=question,
        full_input=messages,
        model_used=model,
    )
    return "".join(block.text for block in message.content if block.type == "text")

Anthropic prompt-cache tokens are counted as input tokens.

Receive control commands

ACP sends signed commands to <Deployed URL>/api/acp/webhook (set the Deployed URL on the app page). Pass the raw request body (bytes), not re-serialised JSON: the signature covers the exact bytes. webhook_response returns the HTTP status (200 applied, 409 for another agent's command, 401 otherwise) and a JSON body.

FastAPI

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

app = FastAPI()

@app.post("/api/acp/webhook")
async def acp_webhook(request: Request) -> JSONResponse:
    status, body = acp.webhook_response(request.headers, await request.body())
    return JSONResponse(body, status_code=status)

Flask

from flask import Flask, jsonify, request

app = Flask(__name__)

@app.post("/api/acp/webhook")
def acp_webhook():
    status, body = acp.webhook_response(request.headers, request.get_data())
    return jsonify(body), status

Django

from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

@csrf_exempt  # ACP requests are authenticated by their HMAC signature, not a CSRF token
@require_POST
def acp_webhook(request):
    status, body = acp.webhook_response(request.headers, request.body)
    return JsonResponse(body, status=status)

No public URL (local development, a worker, a cron job)? Control still works: pause, stop, resume and model swaps arrive with the next heartbeat (within 30 s), and prompt changes are fetched before each run.

API

Security

• Never commit ACP_API_KEY or ACP_WEBHOOK_SECRET; keep them in your platform's secrets.
• ACP_DISABLE_CONTROL=1 (or disable_control=True) ignores every control command while reporting keeps working: a local kill switch that does not depend on ACP.
• Rotate the webhook secret in ACP, then set the old one as webhook_secret_prev until every instance has the new one.
• Commands are accepted only within 5 minutes of their timestamp and only once (nonce cache, safe across threads).
• Run content (user input, full input, the answer) is sent to ACP by default, cut to 100,000 characters per field. Leave out anything you must not share, or use capture_content=False for metadata only.
• Redirects are never followed: requests carry the API key, so a redirect is logged with where ACP moved ("ACP moved to ...; update your url") and the request fails open.
• TLS certificates are verified with the system's default trust store.

Troubleshooting

Warnings are logged to the agent_control_panel logger and reach stderr unless you configure logging.

Forking servers (gunicorn with --preload, multiprocessing). Threads do not survive fork(). The SDK resets itself in the child and restarts its threads on the next report, and runs buffered before the fork stay with the parent. Creating the client in each worker (e.g. gunicorn's post_fork hook) is simplest.

Serverless and short scripts. Call acp.stop() (or use with create_client(...) as acp:) before the function returns so buffered runs are sent; frozen containers do not run exit hooks.

Development

python -m venv .venv && . .venv/bin/activate
pip install --require-hashes -r requirements-dev.txt
ruff check src tests && ruff format --check src tests && mypy && pytest
python -m build

The SDK is developed in the Agent Control Panel repository, next to the server it talks to, so contract changes are tested on both sides together (tests/contract/vectors.json is generated there). This public repository receives one commit per release. Issues and pull requests are welcome here; accepted changes are applied upstream and ship in the next release.

License

Apache License 2.0. It includes an explicit patent grant from contributors, and ends that grant for anyone who sues over patents in the SDK.