Metadata-Version: 2.4
Name: cartha-sdk
Version: 0.4.1
Summary: Cartha SDK — memory, traces, costs, and policies for AI agents
Author-email: Maulik Jadav <maulikjadav239@gmail.com>
Maintainer-email: Maulik Jadav <maulikjadav239@gmail.com>
License-Expression: MIT
Project-URL: Homepage, https://cartha.in
Project-URL: Documentation, https://cartha.in
Project-URL: Repository, https://github.com/maulik-jadav/nexus
Project-URL: Issues, https://github.com/maulik-jadav/nexus/issues
Project-URL: Source, https://github.com/maulik-jadav/nexus/tree/main/nexus/packages/sdk-python
Keywords: cartha,agents,llm,observability,tracing,memory,ai
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx>=0.27
Dynamic: license-file

# 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](https://cartha.in).

## 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](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

```bash
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:

```bash
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:

```python
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 ""
```

| Helper | What it auto-records |
|--------|----------------------|
| `@cartha.trace(...)` | Agent register, heartbeat, run start/finish, nest, optional budget |
| `@cartha.tool()` | `tool_call` steps (success + failure) |
| `cartha.wrap_openai()` | `llm_call` + token cost (feeds budgets) |
| `remember` / `recall` | Scoped memory (still explicit — you choose what to store) |

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`:

```python
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):

| Scope   | Who can recall it                              |
|---------|------------------------------------------------|
| `agent` | only this agent, for this user (private)       |
| `user`  | any of your agents serving this user           |
| `team`  | any agent with the same `team_id`              |
| `org`   | every agent in your organisation               |

```python
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:

```python
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:

```python
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:

```python
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:

```python
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.
```

```python
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:

```python
@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:

```python
@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:

```python
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`:

```python
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:

```python
@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":

```python
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:

```python
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:

```python
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

| Exception            | Raised when                                              |
|----------------------|----------------------------------------------------------|
| `BudgetExceeded`     | the run's cost ceiling was crossed (`.budget_usd`, `.spent_usd`) |
| `ToolNotAuthorized`  | a tool call fell outside the active grant (`.tool_name`, `.allowed_tools`) |
| `PolicyViolation`    | a policy blocked the action (`.policy_name`, `.reason`, `.action`) |

All three are importable from `cartha`.

## From source (development)

```bash
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):

```bash
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.
