Contro1 Python SDK

Use the integration skill

Copy this skill link into your code agent to add Python SDK and Contro1 to your system.

Copy skill link

Key takeaways

• Install the SDK with pip install centcom.
• Use create_protocol_request when an action must pause for human approval.
• Use request_type input when an agent needs a free-text answer from a human.
• Use log_action for allowed autonomous actions that still need durable evidence.
• Verify signed webhooks before resuming any production action.

Install

Use the general Python SDK when your agent, service, worker, tool runner, or webhook bridge is written in Python. It is not tied to Claude Code or any other framework. The same client handles approval and human-input requests, audit records, read APIs, and webhook verification helpers. Runtime-specific connectors are thin adapters on top of the same API.

pip install centcom

CENTCOM_API_KEY=cc_live_your_key
CENTCOM_BASE_URL=https://api.contro1.com/api/centcom/v1
CENTCOM_WEBHOOK_SECRET=whsec_your_secret

Initialize the client

import os
from centcom import CentcomClient

client = CentcomClient(
    api_key=os.environ["CENTCOM_API_KEY"],
    base_url=os.environ.get("CENTCOM_BASE_URL"),
)

Run without an API key

An API key is the quickest start, but a long-lived secret in an environment variable is also the easiest thing for an agent, a log or a leaked image to expose. For agents that run unattended, use an owner-approved connection instead. The client takes exactly one identity: an API key, an owner-approved connection, or the Contro1 service on the same computer.

• On a computer connected with contro1 connect, pass transport=broker_transport(endpoint). The Contro1 service holds the key, so your process holds no credential at all. Unix sockets are supported; on Windows use the JavaScript SDK or the CLI.
• In the cloud (GitHub Actions, Kubernetes, any OIDC issuer), exchange the platform's own identity token with exchange_workload_token. An administrator writes a trust policy and the agent's owner approves it; nothing your code sends decides which agent it becomes.
• Tokens last five minutes and are bound to a key your process or the service holds, so a copied token is useless elsewhere.
• Install the extra for these features: pip install "centcom[runtime]".

import os
from centcom import CentcomClient
from centcom.runtime.auth import broker_transport
from centcom.runtime.dpop import DpopKey
from centcom.runtime.enrollment import exchange_workload_token

# On a connected computer: the local Contro1 service holds the credential
client = CentcomClient(transport=broker_transport(os.environ["CONTRO1_BROKER_ENDPOINT"]))

# In CI or Kubernetes: trade the platform's OIDC token for a 5-minute Contro1 token
key = DpopKey.generate()
tokens = exchange_workload_token("https://api.contro1.com", os.environ["OIDC_TOKEN"], key)
print(tokens["agent_id"], tokens["expires_in"])

Say who can instruct this agent

Contro1 refuses a personal account to an agent that more than one person can instruct. It has to: the agent performs the action with its own authority, so on a shared surface it cannot tell its owner from anybody else who reached it the same way. Put an assistant that reads its owner's mailbox behind an API your customers call, and any of them can ask it what the mailbox holds.

On a platform Contro1 connects to, an adapter reports this. An agent you built with the SDK has no adapter to ask: a nightly job only you trigger and a service answering thousands of customers look identical from here. So you say which it is.

The declaration can only ever make this agent stricter. It arrives on the agent's own credential, which means it is the software describing itself, and letting it claim to be private would let any agent unlock personal accounts by saying so. The server refuses that shape by name. Privacy is established by a person, with contro1 connect, on a machine they control.

Nothing changes for an agent that does not declare it, and nothing changes for organization or shared connections, which already sit inside a resource boundary somebody approved. This is specifically about one person's account being borrowed by software that answers to several.

from centcom import CentcomClient

client = CentcomClient(
    api_key=os.environ["CENTCOM_API_KEY"],
    # Sent once, before the first call. Only ever makes this agent stricter.
    reach={
        "contexts": [
            {
                "context_id": "support-api",
                "label": "Customer support API",
                "kind": "shared",
                "participants_known": False,
            }
        ]
    },
)

Create an approval request

Call create_protocol_request at the point where a risky action is ready but has not executed yet. Use external_request_id for idempotency and correlation_id to group the full business case.

Build context at the gate: copy the exact tool input your code intercepted (machine-observed), attach what triggered the run, and pass the agent's justification by making reason a required parameter of the risky tool. Label model-written text as agent-reported - it informs the reviewer but never changes routing or risk_level.

def send_payment(vendor: str, amount_usd: int, invoice: str, reason: str):
    # "reason" is a required tool parameter: the model justifies at decision time
    request = client.create_protocol_request({
        "title": "Approve vendor transfer?",
        "request_type": "approval",
        "source": {"integration": "finance-agent", "workflow_id": "vendor-payment", "run_id": run_id},
        "routing": {"required_role": "finance", "priority": "urgent", "sla_minutes": 10},
        "context": {
            "action_type": "send_payment",
            "resource": f"vendor:{vendor}",
            "tool_input": {"vendor": vendor, "amount_usd": amount_usd, "invoice": invoice},
            "summary": "New vendor bank account. Invoice INV-9821. Amount $52,400.",
            "agent_reported": {"justification": reason},
        },
        "risk_level": "high",
        "policy_trigger": "Payments above $10,000 require finance approval.",
        "approval_comment_required": True,
        "continuation": {"mode": "decision", "webhook_url": "https://agent.example.com/webhooks/contro1"},
        "external_request_id": f"vendor-payment:{run_id}:atlas-transfer",
        "correlation_id": f"case_vendor_payment_{run_id}",
    })

Ask a human for input

Use request_type input when the workflow needs information rather than permission. The human response is required and returns as decision_type respond. This is different from an optional or policy-required comment attached to Approve or Reject.

request = client.create_protocol_request({
    "title": "Which region should this deployment use?",
    "request_type": "input",
    "source": {"integration": "deployment-agent", "run_id": run_id},
    "context": {
        "action": {"tool": "configure_deployment", "input": {"service": "billing-api"}},
        "machine_observed": {"available_regions": ["us-east-1", "eu-west-1"]},
    },
    "continuation": {"mode": "instruction"},
})

response = client.wait_for_protocol_response(request["request_id"])
if response["decision_type"] != "respond":
    raise RuntimeError("Human input was not provided")

region = response["structured_response"]["value"]

Verify the webhook

Webhook-first is the production path. Verify the signature and timestamp before trusting status, reviewer comments, or structured response data.

from centcom import verify_webhook

ok = verify_webhook(raw_body, signature, timestamp, os.environ["CENTCOM_WEBHOOK_SECRET"])
if not ok:
    raise PermissionError("Invalid Contro1 webhook signature")

if payload["status"] != "approved":
    abort_action()
else:
    resume_action(payload["response"])

Log autonomous actions

Approval requests already store the human decision. Use log_action for actions that were allowed to run automatically but still need searchable evidence.

client.log_action(
    action="report.archived",
    summary="Moved quarterly report from staging to approved archive folder.",
    source={"integration": "document-agent", "workflow_id": "archive-flow", "run_id": run_id},
    outcome="success",
    severity="info",
    correlation_id=case_id,
)

Fetch evidence

evidence = client.get(f"/requests/{request_id}/evidence")

timeline = client.get(f"/cases/{case_id}")

Sub-agents and traces (1.6.0)

client.as_sub_agent(name) is the same agent acting as one named part of it: every call from that view carries the Contro1-Sub-Agent header, and Contro1 registers the part under this agent the first time it is seen. A part runs on this agent's grants and never more.

centcom.tracing.TraceRun reports a run step by step under one trace_id, from the code around the model. run.wrap(tool) reports every call of a tool; run.sub_agent(name) is a part, as a child run. With fail_closed=True a tool whose start could not be recorded does not run.

Ready-made hooks: centcom.hooks.langchain.Contro1TraceHandler (LangChain, LangGraph), centcom.hooks.openai_agents.Contro1RunHooks and centcom.hooks.strands.Contro1HookProvider. Each imports its framework only when used.

researcher = client.as_sub_agent("researcher")
researcher.log_action(action="tool.search", summary="Searched the KB", source={"integration": "crewai"})
researcher.actions.invoke("gmail.message.list", {"max_results": 5}, "agent_principal", "shared")

from centcom.tracing import TraceRun

run = TraceRun(client, source="custom", fail_closed=True)
lookup_order = run.wrap(lookup_order)        # start and end of every call reported

with run.tool("issue_refund", {"order_id": "1842", "amount": 240}):
    issue_refund("1842", 240)

print(run.trace_id)

from centcom.hooks.langchain import Contro1TraceHandler

handler = Contro1TraceHandler(client, fail_closed=True)
graph.invoke(inputs, config={"callbacks": [handler]})

Traceability and sub-agents in full

Send full agent traceability

Beyond the approval call, attach a run trace, the tools you invoked, and the context you retrieved. Each field is optional; add what you have. Identity always comes from your credential. When a named part of a multi-agent system made the call, say which with the Contro1-Sub-Agent header (as_sub_agent in the SDKs): Contro1 registers it under this agent, and it never has more authority than this agent.

• trace_id / parent_trace_id - link one run (and sub-agent runs) into a single trace.
• tool_calls[] - what the agent tried to do, so reviewers see the actions. A hook can report every tool call for you.
• retrieved_context[] - the data the decision was based on (RAG provenance).
• Then export a signed evidence packet from GET /requests/:id/evidence.

# Same approval call you already make - now with full traceability.
billing = client.as_sub_agent("billing")   # optional: a named part of this agent
billing.create_protocol_request({
    "request_type": "approval",
    "title": "Refund $4,200 to customer 8831",
    "source": {"integration": "api"},
    "trace_id": f"trc_{run_id}",          # link every step of this run
    "tool_calls": [{"name": "lookup_order", "outcome": "success"}],
    "retrieved_context": [{"source": "policy:refunds", "uri": "kb://policy/refunds"}],
    "continuation": {"mode": "decision"},
})

Agent identity, traceability & signed evidence

Make the approval the gate, not a suggestion

The signed webhook is cryptographic proof of a human decision. Verify it inside the system that executes the action - not inside the agent. Any tool that must never run without human sign-off (payments, deploys, data deletion) should refuse to act without a verified approval; that way no agent, including shadow agents nobody registered, can trigger it by skipping Contro1.

• Verify the signature and reject timestamps older than 5 minutes (replay protection; the timestamp marks callback delivery, not request creation, so long SLAs are unaffected).
• Bind the approval to the exact action parameters via metadata / correlation_id - never treat "an approval arrived" as permission for a different action.
• Execute each request_id exactly once (idempotency on your side).
• When in doubt, confirm state directly with GET /v1/requests/:id using a read-only API key.

Full guardrail pattern with code: Webhooks

Frequently asked questions

Do I need an API key to use the Python SDK?

No. An API key is one of three identities. On a computer connected with contro1 connect, use broker_transport so the local Contro1 service holds the credential. In CI or Kubernetes, exchange the platform's OIDC token with exchange_workload_token under an owner-approved trust policy. Both avoid a long-lived secret in your environment.

When should I use the Python SDK instead of a framework connector?

Use the Python SDK for custom Python services, workers, and agents. Use a framework connector when you are inside LangGraph, CrewAI, OpenAI Agents SDK, or another supported runtime with its own pause/resume pattern.

Do approval requests automatically create evidence?

Yes. The request stores context, routing, decision, reviewer, timestamps, callback state, and protocol response. Use audit records for additional autonomous or post-approval events.

Related resources