> ## Documentation Index
> Fetch the complete documentation index at: https://mirrors.dev/docs/llms.txt
> Use this file to discover all available pages before exploring further.

# Python quickstart

> Add capture, context retrieval, and completion hooks to your existing Python agent.

Keep your model/tool loop. Add Mirrors where your harness records history,
assembles the next model request, and finishes the turn.

## 1. Install and configure

The Python SDK requires **Python 3.11 or later**. Source version **0.1.0** is
available in the Mirrors repository and is not yet published to PyPI. From a
checkout containing the SDK:

```sh theme={null}
python -m pip install ./packages/python-sdk
```

If you need source access, contact [devs@runmirrors.com](mailto:devs@runmirrors.com).
In the [dashboard](https://platform.runmirrors.com), open your agent's
**Agent setup → SDK connection settings**. Store its API origin, generated agent
ID, and SDK key as `MIRRORS_API_URL`, `MIRRORS_AGENT_ID`, and `MIRRORS_API_KEY`.
The key needs `runtime:read` and `runtime:write`; keep it in server secrets.

Create one client per agent configuration and reuse it across turns. Construct
the client inside each worker process if your application forks:

```python theme={null}
import os
from mirrors_sdk import Mirrors

mirrors = Mirrors(
    api_url=os.environ["MIRRORS_API_URL"],
    api_key=os.environ["MIRRORS_API_KEY"],
    agent_id=os.environ["MIRRORS_AGENT_ID"],
)
```

## 2. Capture history and prepare the model request

At user-turn entry, give the execution a fresh ID. Map your full original history,
including the base system prompt, into [canonical messages](https://mirrors.dev/docs/guides/trace-capture.md).
Keep this ID through that turn's model calls, tools, and retries.

Here, `full_original_history`, `request`, `model.complete`, and `cancel` belong to
your application. `cancel` is its `threading.Event`; preserve your existing
cancellation handling. `request["system"]` contains the original instructions.

```python theme={null}
from concurrent.futures import CancelledError
from uuid import uuid4

original_trace = {"id": str(uuid4()), "messages": full_original_history}
mirrors.capture_trace(original_trace)

selected = mirrors.get_context(original_trace, timeout_ms=1000, cancel=cancel)
if cancel.is_set():
    raise CancelledError()

base_system = request["system"]
response = model.complete(**{
    **request,
    "system": f"{base_system}\n\n{selected.context}" if selected.context else base_system,
})
# Continue your existing response, tool, and stream handling.
```

Capture returns immediately with a ticket. Retrieval sends its own snapshot and
waits at most 1,000 ms, subject to scheduling delays; it never waits for capture.
Empty context is normal in shadow/off mode, on no match, or during a fallback.
The original request and stored system prompt stay unchanged.

Use the returned text immediately for that next model call. Later tool steps use
the base prompt unless you retrieve fresh context for their updated history;
each retrieval adds another bounded wait. Capture updated original snapshots at
your existing history logging points, without injected context.

In an async harness, use `await mirrors.aget_context(original_trace, timeout_ms=1000)`
and your existing async model call. Task cancellation propagates as
`asyncio.CancelledError`; do not catch it and continue to the model.

## 3. Finalize the whole turn

After all model calls, tools, and response streaming finish, send the final
original history under the same ID:

```python theme={null}
ticket = mirrors.capture_trace(
    {**original_trace, "messages": completed_original_history},
    final=True,
    terminal="completed",
)
```

On failure or cancellation, capture the available originals with `terminal="failed"`
or `"canceled"` in your existing ending hook. If no final evidence is available,
use `mirrors.release_trace(original_trace["id"])`. The next turn needs a new ID;
use the [conversation fields](https://mirrors.dev/docs/guides/trace-capture.md#conversation-identity) to group
turns in a continuing chat. Canonical dictionary keys stay camelCase in Python.

Inspect `ticket.receipt` outside model execution, or call `ticket.wait(timeout=2)`
for a wait of up to **2 seconds** (`await ticket.wait_async(timeout=2)` for async).
An accepted capture with ingestion `queued` means analysis is pending.

At shutdown, stop admitting turns and let active turns finish. Call
`delivery = mirrors.close(timeout_ms=2000)` to drain and dispose. Inspect its
`dropped`, `unconfirmed`, and `pending` counts; Python's `delivery.complete` is
true only if all covered tickets were accepted. Async callers can
`await mirrors.aflush(timeout_ms=2000)` then `mirrors.dispose()`.

Start in [shadow mode](https://mirrors.dev/docs/guides/runtime-modes.md) and confirm final conversations appear
in Mirrors. Then verify an active correction reaches the outgoing model request
while stored history stays original. Exercise tool loops, streaming, cancellation,
and an unreachable API. See the [Python reference](https://mirrors.dev/docs/sdk/python-reference.md) for
results and optional metadata, or [troubleshooting](https://mirrors.dev/docs/guides/troubleshooting.md).


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.