# Python SDK (/sdk/python)



The Python SDK (`zumik`) is a synchronous client backed by `httpx`. It wraps the OpenAI-compatible
`/v1` responses surface and the native `/v2` state, diagnostics, and purge endpoints.

## Install [#install]

```bash
pip install zumik
```

Requires Python 3.9+. The only dependency is `httpx`.

## Authenticate [#authenticate]

The API key is an explicit constructor argument; there is no env-var fallback. The base URL defaults
to `https://api.zumik.ai` and the request timeout to 30 seconds.

```python
from zumik import ZumikClient

client = ZumikClient(api_key="zk_live_...")
# or override the defaults:
# ZumikClient(api_key="zk_live_...", base_url="https://api.zumik.ai", timeout=30.0)
```

<Tip>
  Read the key from the environment yourself - the SDK does not: `ZumikClient(api_key=os.environ["ZUMIK_API_KEY"])`.
</Tip>

## /v1 OpenAI-compatible responses [#v1-openai-compatible-responses]

```python
r = client.create_response(model="code.balanced", input="Review the latest patch.")
client.retrieve_response(r["id"])
client.cancel_response(r["id"])
```

For the full OpenAI surface (chat completions, embeddings, streaming), point the official `openai`
package at the Zumik base URL - see [Using the OpenAI SDK](/integrations/openai-sdk) and the
[/v1 responses reference](/api-v1/responses).

## Native /v2 state [#native-v2-state]

Methods are flat on the client and return parsed JSON dicts.

```python
art = client.create_artifact("policy", "Run the linter before every commit.")
bundle = client.create_bundle(items=[{"artifact_id": art["id"], "role": "system"}])
session = client.create_session(base_bundle_ids=[bundle["id"]])

resp = client.create_native_response(
    model="code.fast",
    input="Summarize the failing test.",
    session_id=session["id"],
)
print(resp["output_text"], resp["qos_outcome"])
```

### Branches with compare-and-swap [#branches-with-compare-and-swap]

`append_event` takes the CAS guards as keyword-only arguments:

```python
branch = client.create_branch(session["id"], fork_from_branch_id=session["default_branch_id"])
event = client.append_event(
    session["id"], branch["id"],
    expected_version=branch["version"],
    expected_head_event_id=branch["head_event_id"],
    event_type="user_message",
    payload_ref=art["id"],
)
client.create_snapshot(session["id"], branch["id"], ordered_block_manifest=[art["id"], event["id"]])
```

A stale `expected_version` raises `ZumikApiError` with `status_code == 409` and a
`branch_version_conflict` payload - see [Sessions and branching](/guides/sessions-and-branching).

### Diagnostics, purge, token counts [#diagnostics-purge-token-counts]

```python
diag = client.run_diagnostic(traces=[{
    "trace_id": "trc_1", "privacy_mode": "metadata", "prefix_family_id": "pf_a",
    "observed": {"resolved_target": "openai/gpt-4o@2025-01-01", "ttft_ms": 410, "latency_ms": 2880,
                 "input_tokens": 18000, "candidate_reuse_tokens": 17000,
                 "realized_reused_tokens": 15000, "output_tokens": 400, "attempt_count": 1}}])

job = client.create_purge_job(artifact_ids=[art["id"]])
receipt = client.purge_receipt(job["id"])

client.token_count(input="How many tokens is this sentence")
```

## Methods [#methods]

| Method                                                                                                                | Endpoint                          |
| --------------------------------------------------------------------------------------------------------------------- | --------------------------------- |
| `create_response(model, input)`                                                                                       | `POST /v1/responses`              |
| `retrieve_response(response_id)`                                                                                      | `GET /v1/responses/{id}`          |
| `cancel_response(response_id)`                                                                                        | `POST /v1/responses/{id}/cancel`  |
| `create_artifact(artifact_type, content, **kwargs)`                                                                   | `POST /v2/artifacts`              |
| `create_bundle(items, bundle_type="agent_prefix")`                                                                    | `POST /v2/bundles`                |
| `create_session(base_bundle_ids=None)`                                                                                | `POST /v2/sessions`               |
| `create_branch(session_id, fork_from_branch_id, fork_from_event_id=None)`                                             | `POST /v2/sessions/{id}/branches` |
| `append_event(session_id, branch_id, *, expected_version, event_type, expected_head_event_id=None, payload_ref=None)` | `POST .../events`                 |
| `create_snapshot(session_id, branch_id, ordered_block_manifest, prompt_compiler_revision="pc_1")`                     | `POST .../snapshots`              |
| `create_native_response(model, input, session_id=None, branch_id=None)`                                               | `POST /v2/responses`              |
| `run_diagnostic(traces)`                                                                                              | `POST /v2/diagnostics`            |
| `create_purge_job(artifact_ids)`                                                                                      | `POST /v2/purge-jobs`             |
| `purge_receipt(purge_job_id)`                                                                                         | `GET /v2/purge-jobs/{id}/receipt` |
| `token_count(input)`                                                                                                  | `POST /v2/token-counts`           |

## Error handling [#error-handling]

Any HTTP status ≥ 400 raises `ZumikApiError`, carrying `status_code` and the parsed `payload`:

```python
from zumik import ZumikClient, ZumikApiError

try:
    client.create_artifact("policy", "")
except ZumikApiError as e:
    print(e.status_code, e.payload["error"]["code"])
```

The SDK does not retry or stream. For OpenAI-style streaming, use the `openai` package against the
`/v1` base URL.