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