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.

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.

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, tool_call() raises ToolNotAuthorized before the call is recorded as having happened — 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. 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>.

12. 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.0).

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.