Cartha SDK (Python)

The memory + accountability layer for AI agents — scoped memory, full trace replay, time-travel recall, budget breakers, and policy guardrails, from one decorator. Built by Cartha.

Before you install: create your account

The SDK talks to your Cartha workspace, so you need an API key first:

1. Sign up / log in at https://cartha.in/login (Google or GitHub — takes ~30 seconds).
2. Open Settings in the dashboard and copy your API key and API base URL.

Without these two values the SDK has nowhere to send anything — do this first.

Install

pip install cartha-sdk
# optional — for automatic LLM tracing:
pip install openai

Requires Python 3.10+.

Then configure the two values from your dashboard Settings page:

export CARTHA_API_KEY="cartha_..."          # dashboard → Settings → API key
export CARTHA_API_BASE="https://cartha.in"  # dashboard → Settings → API endpoint

Easy path (recommended)

Most of a full integration with almost no boilerplate:

import cartha

cartha.init()
client = cartha.wrap_openai()  # auto llm_call + cost on every chat completion

@cartha.tool()                 # auto tool_call success/failure
def crm_lookup(user_id: str) -> dict:
    return {"plan": "pro"}

@cartha.trace(id="support_agent", team="support", budget_usd=0.50)
def handle(user_id: str, ticket: str) -> str:
    cartha.remember_sync(user_id=user_id, content=ticket, scope="user")
    hits = cartha.recall_sync(user_id=user_id, context=ticket, scope=["user", "team"])
    data = crm_lookup(user_id)
    r = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"{ticket}\n{data}\n{hits}"}],
    )
    return r.choices[0].message.content or ""

Run it, then open the dashboard — the run is there, step by step, including the exception and stack trace if it failed. @cartha.trace() works on both sync and async functions. See examples/easy_full_agent.py.

Zero-code auto-instrumentation (optional)

If you already use OpenAI, LangGraph, Anthropic, Gemini, LiteLLM, CrewAI, Pydantic AI, AutoGen, LlamaIndex, or similar, call instrument() once after init() to monkey-patch installed libraries:

import cartha

cartha.init()
cartha.instrument()              # patch every supported library that is installed
# cartha.instrument(["openai"])  # or only specific frameworks

What auto-patching actually does today (honest matrix — prefer @trace / @tool / wrap_openai for production governance):

Hard budgets, allowed_tools, and named agents still need @cartha.trace(...). LangGraph tools are allow-list gated before the tool body runs (same contract as @cartha.tool). LLM instrumentors estimate total_cost_usd from token counts so budget breakers can trip.

For a single LangGraph app instead of process-wide patching:

app = cartha.observe(graph.compile())
app.invoke({"messages": [...]})

CLI

cartha doctor      # Python, API key, /health connectivity, installed frameworks
cartha version
cartha config      # whether CARTHA_API_KEY / CARTHA_API_BASE are set
cartha instrument python app.py   # run a script with auto-instrument env

Full product docs: https://cartha.in/documentation

Everything below builds on this. The full instrumentation API lives in cartha.ops:

from cartha import ops

Feature guide

1. Scoped memory — remember & recall

Agents store and retrieve memories with an explicit scope that is enforced server-side at three layers (vector filter, SQL, application):

from cartha import ops

@cartha.trace(id="support_agent", team="support")
async def handle_ticket(user_id: str, question: str):
    # Store a fact about this user (visible to all your agents serving them)
    await ops.remember(
        user_id=user_id,
        content="Customer prefers email over phone contact.",
        scope="user",
        confidence=0.9,
        decay_days=30,          # expires automatically
    )

    # Retrieve relevant memories (semantic search + scope enforcement)
    hits = await ops.recall(
        user_id=user_id,
        context=question,        # what you're trying to answer
        scope=["user", "team"],  # which tiers this agent may read
        top_k=5,
    )
    for h in hits:
        print(h["memory"]["content"], h["score"])

Scope isolation is real, not decorative: agent B recalling with scope=["agent"] can never see agent A's private memories, and a different team can never see your team-scoped ones.

Default agent registration grants writable user / agent / team (not org until you configure it). Pass an explicit scope= on every remember / recall — never rely on “whatever the agent can see.”

2. Time-travel recall — what did the agent know then?

The debugging question that's normally unanswerable: "why did it say that last Tuesday?" Pass as_of and recall returns the memory state exactly as it stood at that instant — memories added since are excluded, memories that had already expired then are excluded:

hits = await ops.recall(
    user_id="u1",
    context="refund policy",
    scope=["user"],
    as_of="2026-07-12T14:30:00Z",   # or a datetime object
)

Pass a trace's timestamp from the dashboard and the mystery becomes a deterministic replay. (Erased memories never resurface — even in time travel. Compliance wins.)

3. Tool & LLM call tracing

cartha.wrap_openai() and @cartha.tool() (see the easy path above) record these automatically. To instrument manually — other model providers, custom tools — record every call so the dashboard shows the full decision path, latency, and cost of each run:

await ops.tool_call(
    tool_name="crm_lookup",
    input_schema={"invoice_id": "8492"},
    output={"status": "found"},
    latency_ms=210,
)

await ops.llm_call(
    model="claude-sonnet-5",
    input="Summarize this ticket...",
    output="The customer wants...",
    tokens_in=420, tokens_out=180,
    latency_ms=900,
    total_cost_usd=0.0031,   # feeds cost tracking AND the budget breaker
)

Or record spend directly:

await ops.cost(model="claude-sonnet-5", tokens_in=420, tokens_out=180,
               total_cost_usd=0.0031)

4. Budget breaker — stop the $300 loop

A hard cost ceiling per run. When the limit is crossed, the very next spend raises BudgetExceeded — the runaway loop stops itself instead of showing up on your invoice:

import cartha

@cartha.trace(id="researcher", team="ops", budget_usd=5.00)
async def research(user_id: str, topic: str):
    for source in sources:                      # imagine this loops forever
        await ops.llm_call(model="...", input=..., output=...,
                           total_cost_usd=0.03)
    # When cumulative spend crosses $5.00 → BudgetExceeded is raised here,
    # on the next spend — overshoot is bounded to roughly one call.

try:
    await research(user_id="u1", topic="...")
except cartha.BudgetExceeded as e:
    print(e.budget_usd, e.spent_usd)  # 5.0, 5.02

Enforcement is layered: a fast local counter (microseconds, no network hop in your hot path) plus an authoritative server-side ledger, so even a multi-process agent fleet can't spend past the cap by resetting a local counter. Set a default via CARTHA_BUDGET_USD env if you prefer.

5. Tool allow-lists — hard authority boundaries

Restrict which tools a run may call. Outside the list, assert_tool_allowed (used by @cartha.tool and LangGraph BaseTool auto-instrumentation) raises ToolNotAuthorized before the tool body runs — a boundary, not an after-the-fact audit finding:

@cartha.trace(id="intern_agent", team="ops",
              budget_usd=2.00, allowed_tools=["search", "summarize"])
async def intern(user_id: str):
    await ops.tool_call(tool_name="search", output="...")        # fine
    await ops.tool_call(tool_name="wire_transfer", output="...") # raises ToolNotAuthorized

6. Attenuated delegation — parent grants child a subset

The multi-agent problem: a parent hands work to a child without handing over its whole budget or tool authority. delegate() mints a child grant carved out of the parent's remaining balance — atomic, so two children can never be sold the same dollar, and the child's tool list can never be wider than the parent's:

@cartha.trace(id="orchestrator", team="ops",
              budget_usd=10.00, allowed_tools=["search", "email"])
async def orchestrator(user_id: str):
    grant = await cartha.delegate(
        to_agent_id="worker",
        task_description="handle the sub-task",
        budget_usd=2.00,              # child gets $2 of the parent's $10
        # allowed_tools omitted → child inherits ["search", "email"] exactly.
        # Requesting ["search", "wire_transfer"] would be REJECTED (422) —
        # a child can never escalate beyond the parent's grant.
    )
    try:
        return await worker(
            user_id=user_id,
            cartha_budget_id=grant["budget_id"],
            cartha_budget_max_usd=grant["budget_max_usd"],
            cartha_budget_tools=grant["budget_allowed_tools"],
        )
    finally:
        if grant.get("budget_id"):
            # Release the child's unspent balance back to this budget.
            await cartha.close_budget(grant["budget_id"])

@cartha.trace(id="worker", team="ops")
async def worker(user_id: str, **cartha_ctx):
    # This run is clamped to the $2 / ["search", "email"] grant.
    # Spending $2.01 raises BudgetExceeded; calling another tool raises
    # ToolNotAuthorized — regardless of what "worker" is normally allowed.
    ...

Advanced: ops.open_budget(max_usd=...) / ops.close_budget(budget_id) manage envelopes directly; closing returns the unspent balance to the parent.

7. Retries that respect the budget

retry_context groups attempts of one flaky operation so the dashboard collapses them into a single entry — and rc.check() stops a retry loop from burning attempts after the budget has already tripped:

rc = ops.retry_context(max_attempts=3)
for attempt in rc:
    rc.check()   # raises BudgetExceeded before wasting another attempt
    try:
        result = await flaky_tool()
        await ops.tool_call(tool_name="flaky", output=result, **rc.step_kwargs())
        break
    except TimeoutError:
        await ops.tool_call(tool_name="flaky", status="timeout", **rc.step_kwargs())

8. Policy guardrails

Write rules in plain English on the dashboard (Policies page) — e.g. "Never share financial information" or "Block the CRM tool unless explicitly requested". The SDK enforces them in-process (the compiled policy bundle is cached and evaluated locally in microseconds; only the ~30s refresh touches the network). A blocked action raises PolicyViolation:

try:
    await ops.remember(user_id="u1",
                       content="Their bank account number is...",
                       scope="user")
except cartha.PolicyViolation as e:
    print(e.policy_name, e.reason, e.action)   # blocked before it was stored

Policies with action require_human_approval create an escalation a human resolves on the dashboard — the agent polls its verdict instead of guessing.

9. Multi-agent tracing (nesting & cross-service)

Nesting is automatic: a traced agent calling another traced agent produces a linked child trace, so multi-agent runs render as a tree, not a blur:

@cartha.trace(id="parent")
async def parent(user_id: str):
    await ops.delegate(to_agent_id="child", task_description="sub-task")
    return await child(user_id)      # auto-nests under parent

@cartha.trace(id="child")
async def child(user_id: str): ...

Crossing a service boundary? Send ops.propagation_context() with the request and set CARTHA_PARENT_TRACE_ID on the other side — the remote trace links back to this one.

10. Cost per completed task

Retries and failed re-runs of one logical job share a task_id, so the dashboard can answer "what did completing this actually cost?" rather than just "what did each attempt cost":

tid = cartha.task_context()
for attempt in range(3):
    try:
        await run_agent(user_id="u1", cartha_task_id=tid)
        break
    except Exception:
        continue   # same tid → all attempts roll up into one outcome

11. Agent-task fit — which agent is actually better at what

Declare the kind of work a run does with task_type. Two agents that declare the same task_type become comparable, and the dashboard ranks them on the blend that matters: success rate first, then cost per run, then latency.

@cartha.trace(id="premium_agent", team="support",
              task_type="invoice_refund", budget_usd=5.0)
async def premium(user_id: str, invoice_id: str): ...

@cartha.trace(id="backup_agent", team="support",
              task_type="invoice_refund", budget_usd=1.0)
async def backup(user_id: str, invoice_id: str): ...

task_type is the reusable category ("invoice_refund"); task_id is one specific job ("ticket_9021"). An agent needs at least 3 runs before it is ranked — a lucky 100% on two runs never outranks a real track record.

See it at GET /api/v1/intelligence/task-fit, or on the Intelligence page.

12. Failover — hand the task on when an agent hits its ceiling

budget_usd stops a runaway agent. run_with_failover() decides what happens next: the task moves to the next agent in the chain instead of just failing.

result = await cartha.run_with_failover(
    task_type="invoice_refund",
    handlers={"premium_agent": premium, "backup_agent": backup},
    user_id="customer_123",
    invoice_id="8401",
)

When premium_agent raises BudgetExceeded at $4.98 of its $5.00 ceiling, backup_agent picks the task up — and because both attempts share one task_id, the memory the first agent stored is already there, and the dashboard shows one job with two attempts, not two unrelated runs.

Order comes from an explicit chain when you configure one, and otherwise from the learned ranking in §11 — so failover works before anyone writes config:

# optional: pin the order yourself
# POST /api/v1/intelligence/routing/rules
#   {"task_type": "invoice_refund", "agent_key": "backup_agent", "priority": 1}

Ordinary exceptions do not fail over by default — retrying a genuine bug on a second agent usually just buys the same crash twice. Pass on_failure=True if you want that too.

13. Prompt version snapshots & run diff

Snapshot the prompt template a step ran with, and the platform content- addresses it — so replay stays faithful after the template changes, and the run-diff endpoint can tell you "these two runs diverged because the prompt changed" instead of blaming a downstream step:

await ops.tool_call(
    tool_name="draft_email",
    output=draft,
    prompt=ops.prompt_snapshot("email_template", template_text),
)

Compare any two runs on the dashboard, or via GET /api/v1/traces/diff?left=<trace>&right=<trace>.

14. No-code platforms (n8n, Make, Zapier, Dify, …)

No SDK, no Python: a no-code workflow can log a whole run in one HTTP call at the end of the flow.

curl -X POST https://api.cartha.in/api/v1/ingest/run \
  -H "X-Api-Key: $CARTHA_API_KEY" -H "Content-Type: application/json" \
  -d '{
    "agent": "support_workflow",
    "platform": "n8n",
    "task_type": "invoice_refund",
    "status": "success",
    "steps": [
      {"type": "tool", "name": "crm_lookup", "input": {"invoice_id": "8401"}},
      {"type": "llm", "model": "gpt-4o-mini", "cost_usd": 0.0021}
    ]
  }'

The agent registers itself on first use, and these runs get the same traces, cost tracking and agent-task-fit comparison as instrumented Python. Field names are forgiving (agent/agent_id/workflow, cost/cost_usd, nodes/steps), and re-POSTing the same trace_id is ignored — so a retrying workflow node can't double-count spend.

GET /api/v1/ingest/template?platform=n8n returns a ready-to-paste HTTP-node config for your platform.

15. Sync code

Every operation has a _sync twin for non-async codebases:

cartha.remember_sync(user_id="u1", content="...", scope="user")
hits = cartha.recall_sync(user_id="u1", context="...", scope=["user"])
cartha.tool_call_sync(tool_name="search", output="...")

Memory denial mode (org setting)

What does an agent see when it asks for a scope it isn't allowed to read?

• denied_hint (default): an explicit, content-free denial — the agent knows context was withheld instead of confidently inventing over a hole.
• silent: empty results only (no existence signal at all).

Set on the dashboard (Settings) or PATCH /api/v1/org.

Exceptions summary

All three are importable from cartha.

From source (development)

pip install "git+https://github.com/maulik-jadav/nexus.git#subdirectory=nexus/packages/sdk-python"

Legacy alias: older code may from nexus import ops. That still works for compatibility, but new code should use cartha.

Publishing (maintainers)

Releases are published from GitHub Actions when you push a tag matching sdk-python-v* (for example sdk-python-v0.4.4).

Manual upload (emergency / first release):

cd nexus/packages/sdk-python
python -m pip install --upgrade build twine
python -m build
python -m twine upload --repository testpypi dist/*   # TestPyPI first
python -m twine upload dist/*                          # production PyPI

Use a PyPI API token (pypi-...), not your account password.