Skip to main content
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:
If you need source access, contact devs@runmirrors.com. In the dashboard, 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:

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. 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.
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:
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 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 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 for results and optional metadata, or troubleshooting.