CogPlane Python SDK

Send HMAC-authenticated agent evidence to CogPlane without placing CogPlane in the agent's critical execution path. The default client queues records locally and fails open if evidence delivery is temporarily unavailable.

Five-minute onboarding

1. Install

python -m pip install cogplane

2. Get an ingestion key

An organization administrator opens Settings → SDK ingestion identity, selects the exact workspace, and chooses Issue ingestion key. The returned cgp_ingest_ value is shown once and carries only cognition.write; save it directly in the workload secret manager. Revoking that ingestion identity disables its service actor and all of its API keys.

For operator-created organizations, an operator can instead create the tenant:

$headers = @{
  Authorization = "Bearer $env:COGPLANE_ADMIN_TOKEN"
  "Content-Type" = "application/json"
}
$body = '{"org_id":"acme","project_id":"invoice-pilot","env":"prod"}'
Invoke-RestMethod -Method Post -Uri "$env:COGPLANE_URL/admin/tenants" `
  -Headers $headers -Body $body

Copy the returned ingest_key into a secret manager and expose it to the agent as COGPLANE_INGEST_KEY. When COGPLANE_PUBLIC_URL and the RBAC store are available, tenant creation also provisions one sdk-default service actor with only cognition.write. Its one-time cpk_ API key is carried inside the returned ingestion key; it is not exposed as a second top-level secret. The non-secret sdk_actor response field identifies the actor and its permission.

An ingestion key starts with cgp_ingest_. The complete token contains both the tenant signing secret and least-privilege RBAC actor key in encoded, not encrypted, form. Treat it exactly like a password: never put it in source code, logs, screenshots, tickets, or browser storage.

GET /admin/tenants/ingest-key deliberately remains side-effect-free and returns a legacy secret-only token; it does not create or return an actor key. On an RBAC-enabled server, obtain the complete one-key token from org-admin Settings or POST /admin/tenants. The legacy operator-only recovery path may pair a reissued token with a key from POST /admin/rbac/actors using CogPlane(key=..., api_key=...).

3. Record an event

import os
from cogplane import CogPlane

cog = CogPlane(
    key=os.environ["COGPLANE_INGEST_KEY"],
    run_id=os.environ.get("COGPLANE_RUN_ID"),  # optional cross-restart continuity
)
cog.record(agent="invoice-bot", input="Approve invoice 42?", output="Escalated")
cog.flush(timeout=5)

run_id is a customer-chosen grouping label, not a server-side workflow. There is no server-side start, stop, completion, owner, or status lifecycle. Without an explicit run_id, one is generated when the SDK client starts and shared by every event from that client instance. A long-lived client can therefore emit one run for months; constructing one client per request creates one run per request. Neither automatically means one business job.

When evidence must follow one business job across workers, retries, process restarts, or deployments, create an opaque run ID in the workload orchestrator, persist it with that job, and pass it to every reporting client. Do not derive a run ID from model conversation memory.

record() returns promptly after enqueueing. Call flush() before a short-lived process exits, or use a context manager so shutdown drains the queue:

with CogPlane(key=os.environ["COGPLANE_INGEST_KEY"]) as cog:
    cog.record(agent="invoice-bot", input="Invoice 42", output="Approved")

The queue is bounded and drops its oldest pending item rather than blocking the agent. Inspect cog.metrics for queue_size, dropped_count, and failed_count. Set strict=True only when evidence delivery must raise back to the caller; this can affect the customer workload.

4. Auto-instrument an existing model client

cog.instrument_openai(openai_client, agent="invoice-bot")
result = openai_client.chat.completions.create(model="gpt-5", messages=messages)
cog.uninstrument_openai(openai_client)

cog.instrument_anthropic(anthropic_client, agent="invoice-bot")
result = anthropic_client.messages.create(
    model="claude-sonnet-5", max_tokens=500, messages=messages
)
cog.uninstrument_anthropic(anthropic_client)

Instrumentation returns the vendor's original object and preserves the vendor's original exception. CogPlane recording failures are logged but do not replace a successful model response. Streaming responses are never consumed; CogPlane records metadata only.

5. Prove which registered agent sent the event

Tenant HMAC authenticates the tenant credential used for an event. For individual agent proof, register the agent in CogPlane and use its one-time agt_ credential:

agent = cog.agent("invoice-bot", key=os.environ["COGPLANE_AGENT_KEY"])
agent.record(input="Approve invoice 42?", output="Escalated")

The server can require this second agent HMAC with COGPLANE_REQUIRE_AGENT_PROOF=true.

Production always requires actor to match an active agent registered in the same tenant. Development and staging retain observe-only compatibility unless COGPLANE_REQUIRE_REGISTERED_AGENTS=true is set. Registration establishes tenant-authorized attribution; the second agent HMAC shows that the sender held the registered agent's shared credential. Because HMAC is symmetric, this is not third-party non-repudiation.

Download an Evidence Pack

Evidence Pack download is a server API operation, separate from asynchronous SDK ingestion. Provision an RBAC actor with export.read, then call GET /evidence/pack with canonical tenant headers and the required tenant, from, and to query fields. Optional repeated agent, department, and include filters scope activity, drift, health, and usage consistently.

The zip contains HMAC-authenticated JSON evidence when keyed, SUMMARY.pdf, VERIFY.md, and a standard-library-only verify.py. Run python verify.py pack.zip offline. Provide --key through a separate approved channel to verify HMAC authenticity; the signing key is never included in the pack. Filtered activity proves per-record integrity and sealed-window membership, not an unbroken chain. See the root API documentation for a complete request example.

Compatibility path

Existing integrations can continue constructing CogPlaneClient with an endpoint, canonical org/project/env tenant, and signing key. That constructor remains strict by default to avoid silently changing legacy behavior. Legacy three-field ingestion tokens also continue to decode byte-for-byte as before. If CogPlane receives both an embedded actor key and an explicit api_key=, the explicit key wins.