Integrate the Python SDK

Use caracalai-sdk in an async Python application that must create governed Sessions, narrow authority, route HTTP through Gateway, or bind verified inbound context. Resource servers that only verify inbound mandates should use the ASGI adapter or verify package.

Prerequisites

A managed application credential, active policy, resource/provider binding, and Gateway route.
An async runtime; move CPU-bound work off the event loop so service heartbeats can run.
A timeout and destination idempotency plan for mutating requests.

Install

pip install caracalai-sdk

Connect

from caracalai import Caracal

caracal = Caracal()

Caracal() loads exactly the profile named by CARACAL_CONFIG when set; otherwise it loads CARACAL_* environment variables. It does not search default profile or credential paths, and conflicting credential modes fail at startup. Use Caracal.from_client_secret(...) for complete static credentials supplied in code. Power-user loaders and resolver-backed credentials live in caracalai.advanced.

Start a Session and call a protected resource

The smallest protected call needs no session code at all: application_transport() pins an httpx.AsyncClient to one resource and calls as the application’s own identity, provisioning the required Session and Delegation for you.

from caracalai import Caracal

caracal = Caracal()
target = caracal.gateway_request("resource://pipernet", "/reports")
async with caracal.application_transport(
    "resource://pipernet",
    scopes=["pipernet:read"],
) as governed:
    response = await governed.get(target.url)

When your code already runs inside a governed Session - an agent step, a worker task - call through the Session instead: fetch() mints a one-shot Gateway mandate under the Session’s authority.

from caracalai import Authority, Caracal

caracal = Caracal()
resource_id = "resource://pipernet"

async with caracal.session() as parent:
    async with caracal.session(
        parent_ctx=parent,
        authority=Authority.narrow(
            ["pipernet:read"],
            resource_id=resource_id,
            ttl_seconds=600,
        ),
    ):
        await caracal.fetch(resource_id, "/reports", scopes=["pipernet:read"])

Why two levels? The outer Session is the lifecycle parent; the narrowed child creates the positive-TTL Delegation that resource authority requires. Both async context managers retire their Sessions on exit. application_transport() builds exactly this structure internally, which is why it is the right starting point for application-owned calls.

To make AI-agent executions distinguishable in policy and audit, pass labels when starting a Session. These become input.principal.labels, so several agents under one application stay separable without one application per agent. Labels describe work; scopes and Delegations bound authority. session() retires the task Session when the block exits, while start_session() starts a heartbeat-leased long-lived Session whose protocol lifecycle is service.

async with caracal.session(labels=["refund-agent"]) as ctx:
    print("refund Session", ctx.session_id)

See If many agents share one managed application, can policy and audit still tell them apart?.

Long-lived Sessions

Daemons and workers that outlive a single request use start_session() instead of session(). It returns a handle you own: the SDK renews the lease from an independent background task by default, and you retire the Session with aclose(). If heartbeats stop before the lease expires, Coordinator suspends the Session for explicit recovery or termination. The stored protocol lifecycle is service; unlike task, it is not subject to the wall-clock TTL sweeper.

svc = await caracal.start_session(labels=["fiona-worker"])
try:
    while running:
        async with caracal.bind(svc.context):
            await do_work(caracal.headers())
finally:
    await svc.aclose()

Narrowing requires an active parent. For a long-lived hierarchy, bind a parent service handle while starting its narrowed service child. The child context carries the edge, and closing each handle retires its own Session.

parent = await caracal.start_session(labels=["pipernet-orchestrator"])
async with caracal.bind(parent.context):
    svc = await caracal.start_session(
        labels=["fiona-worker"],
        authority=Authority.narrow(
            ["pipernet:read"],
            ttl_seconds=600,
            resource_id="resource://pipernet",
        ),
    )

The renewal cadence follows the server lease, renewing at roughly a third of the remaining lease with jitter. Pass a positive heartbeat_interval to fix the cadence, or 0 to disable the background task and call heartbeat() yourself. Because renewal runs on an independent task, the lease stays current even while your code is blocked on a long await (a streaming response, a slow tool).

The handle exposes heartbeat_deadline_at and lease_generation. Every heartbeat and aclose() sends that monotonic ownership token, so a process holding an earlier generation cannot renew or terminate the Session after another process takes ownership.

Transient renewal errors are logged and retried on the next tick rather than crashing the worker. If Coordinator reports the Session permanently gone or the generation has been fenced by a new holder, the task stops and on_lease_lost fires once so the worker can resign instead of spinning. An expired lease instead suspends the Session and reports suspended through on_state_change.

The handle’s status follows Coordinator heartbeat responses. If it becomes suspended, automatic heartbeat stops and on_state_change reports the transition. Stop taking work, resolve the cause, resume the Session through the control plane, and call attach_session() to restart lease renewal. on_lease_lost remains reserved for terminal loss.

svc = await caracal.start_session(
    labels=["voice-worker"],
    on_lease_lost=lambda exc: shutdown(),
)

A renewal cannot run while the event loop is blocked synchronously (CPU-bound work with no await); in that case the lease correctly lapses, which is the liveness signal working as intended.

A worker that restarts does not need a fresh session: persist svc.session_id and re-attach with attach_session(). Attach atomically acquires a new lease generation and renews the deadline, fencing every older handle. A Session the Coordinator no longer holds live fails with CoordinatorError; a suspended Session must be resumed through the control plane before attachment. The returned handle behaves like one from start_session():

svc = await caracal.attach_session(
    persisted_session_id,
    on_lease_lost=lambda exc: shutdown(),
)

The rebuilt context carries the Session identity only; Delegations bound by the previous holder are re-presented with accept_delegation().

Start a narrowed child

from caracalai import Authority, DelegationConstraints

async with caracal.session() as parent:
    async with caracal.session(
        parent_ctx=parent,
        authority=Authority.narrow(
            ["pipernet:read"],
            resource_id="resource://pipernet",
            constraints=DelegationConstraints(max_hops=1),
            ttl_seconds=600,
        ),
    ):
        headers = caracal.headers()

A plain caracal.session() runs the child under its parent’s effective authority - the application’s authority for a root parent, or the parent’s narrowed slice when the parent was itself narrowed (transitive least-privilege). Pass authority=Authority.narrow(...) only when the child should hold a smaller subset, or authority=Authority.none() for a child with no inherited authority. Use await caracal.delegate(...) when you need to grant authority to a Session that already exists, typically in another application: it returns the edge, and the receiving session presents it with async with caracal.accept_delegation(edge_id).

Use async with caracal.bind(ctx) before handing a captured context to a background task.

Handle approvals

A mint whose scope is approval-gated raises ApprovalRequired. with_approval runs the whole flow - catch the hold, wait for the decision, retry with the approval id - in one call. Its callback must return an awaitable; mint_mandate is synchronous, so wrap it with asyncio.to_thread:

import asyncio

mandate = await caracal.with_approval(
    lambda approval_id: asyncio.to_thread(
        caracal.mint_mandate,
        "resource://pipernet",
        ["pipernet:admin"],
        approval_id=approval_id,
    )
)

A rejected, expired, or already-consumed decision re-raises the original ApprovalRequired; its approval id lets you resume waiting later with wait_for_approval, which returns the typed final state (approved, rejected, expired, consumed, or pending). Human Approval covers the tiers and decision paths.

Use httpx transport injection

async with caracal.session():
    async with caracal.transport(
        scopes=["pipernet:read"], propagation="gateway-only"
    ) as client:
        await client.get("https://api.pipernet.example/reports")

The transport mints the scoped Gateway-ingress mandate, injects Caracal envelope headers, and rewrites configured resource-bound URLs through the Gateway. Calls without scopes require an already Gateway-class bound token.

ASGI propagation

After a verifier boundary, use the SDK middleware to bind an inbound Caracal envelope into request context:

from fastapi import FastAPI
from caracalai import Caracal

caracal = Caracal()
app = FastAPI()
app.add_middleware(caracal.context_middleware(trusted_propagation=True))

This propagation-only form is valid only when a Gateway or adapter already verified the request. At an untrusted ingress, pass a complete verifier= instead; production mode rejects propagation that is neither verified nor explicitly trusted.

Troubleshooting

Symptom	Check
`Caracal: missing ...`	Confirm the named profile or required environment variables.
`headers()` refuses root identity	Bind a context or pass `as_application=True` only for trusted service-root calls.
Background task loses context	Capture the context and rebind with `async with caracal.bind(ctx)`.
Gateway routing misses	Confirm resource bindings and `gateway_url`.

Validate the integration

Exercise allow, extra-scope deny, revoked Session, cancellation, and clean shutdown. Only the allow case may reach the upstream. Close httpx transports and owned service handles, then await caracal.aclose() when the client is no longer used.

For exact Python signatures and sync/async variants, use Python SDK reference.

Next Step

Protect inbound FastAPI traffic with the ASGI adapter or implement multi-agent delegation for cross-Session authority.