Integrate the Python SDK

Use caracalai-sdk in Python services and agents that need async session lifecycle, delegation, Gateway routing, or ASGI context propagation.

Install

pip install caracalai-sdk

Connect

from caracalai import Caracal

caracal = Caracal()

Caracal() loads CARACAL_CONFIG, a caracal.toml in the default config path, or environment variables.

Spawn and call a resource

import httpx
from caracalai import Caracal

caracal = Caracal()

async with caracal.spawn():
    headers = caracal.headers()
    async with httpx.AsyncClient() as client:
        await client.get("https://api.example.com/tickets", headers=headers)

spawn() is an async context manager. It binds the agent context inside the block and releases the session when the block exits.

To make agents distinguishable in policy and audit, pass labels when you spawn. These become input.principal.labels, so several agents under one application stay separable without one application per agent. labels are descriptive, for policy and audit, not grants; authority always comes from scopes and delegation. The session lifecycle is handled for you: spawn() retires the session when the block exits and records it as a task, while spawn_service() records a heartbeat-leased service.

async with caracal.spawn(labels=["refund-agent"]):
    headers = caracal.headers()

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

Long-lived service agents

Daemons and workers that outlive a single request use spawn_service() instead of spawn(). It returns a handle you own: keep the session alive by calling heartbeat() on a timer and retire it with aclose(). Each heartbeat() renews the lease — it extends the deadline the coordinator measures liveness against; it does not merely check status. A service session is reaped by the coordinator only if it stops heartbeating before its lease expires. Unlike a task, a service is not subject to the wall-clock TTL sweeper; its lifetime is governed entirely by the lease, so a faithfully heartbeated service runs until you close it.

svc = await caracal.spawn_service(labels=["billing-worker"])
try:
    while running:
        await svc.heartbeat()
        await do_work(caracal.headers())
        await asyncio.sleep(30)
finally:
    await svc.aclose()

Service handles take the same grant= as spawn(): pass Grant.narrow(...) to issue a delegation edge that bounds the service’s authority for its whole lifetime, exactly as a narrowed task child. The handle’s context carries the edge, and closing the handle retires the session.

svc = await caracal.spawn_service(
    labels=["billing-worker"],
    grant=Grant.narrow(["ledger:read"], resource_id="resource://ledger"),
)

When the main coroutine spends long stretches awaiting a provider (a streaming response, a slow tool), renewing the lease from the same loop can starve. Pass heartbeat_interval to renew the lease from an independent background task, so the lease stays current even while your code is blocked on a long await:

svc = await caracal.spawn_service(labels=["voice-worker"], heartbeat_interval=30)
try:
    async for chunk in stream(caracal.headers()):   # lease renews in the background
        await handle(chunk)
finally:
    await svc.aclose()

Auto-heartbeat is opt-in (omit heartbeat_interval to renew manually). It renews even while your code awaits, and transient renewal errors are logged and retried on the next tick rather than crashing the worker. 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.

Spawn a narrowed child

from caracalai import Grant, DelegationConstraints

async with caracal.spawn() as parent:
    async with caracal.spawn(
        parent_ctx=parent,
        grant=Grant.narrow(
            ["tickets:read"],
            resource_id="https://api.example.com/tickets",
            constraints=DelegationConstraints(max_hops=1, budget=5),
            ttl_seconds=600,
        ),
    ):
        headers = caracal.headers()

A plain caracal.spawn() 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 grant=Grant.narrow(...) only when the child should hold a smaller subset, or grant=Grant.none() for a child with no inherited authority.

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

Use httpx transport injection

async with caracal.spawn():
    async with caracal.transport() as client:
        await client.get("https://api.example.com/tickets")

The transport injects Caracal envelope headers and rewrites configured resource-bound URLs through the Gateway.

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())

The middleware propagates context. It does not verify JWT signatures or revocation; use a connector or transport verifier at the trust boundary.

Troubleshooting

Symptom	Check
`Caracal.from_env: missing ...`	Confirm profile or required environment variables.
`headers()` refuses root identity	Bind a context or pass `allow_root=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`.