Skip to main content
This reference covers the Python source package mirrors-sdk 0.1.0, which requires Python 3.11 or later and is not yet published to PyPI. Start with the Python quickstart for installation and integration.

Client and context

Create Mirrors(api_url=..., api_key=..., agent_id=...) once per agent configuration and reuse it across executions. Keep the API key in server secrets. Construct a new client inside each process if your application forks workers. get_context(trace, timeout_ms=1000, cancel=None) is synchronous. It returns a ContextResult with attributes, not a dictionary: Call once before assembling the user turn’s model request. The total wait budget is 1–1,000 ms, including copying and scheduling; it does not wait for capture uploads. Scheduling delays can exceed the budget. accepted confirms a decision arrived, not that your application used it or the model succeeded. Use result.context promptly. Check result.freshness.expires_at if dispatch is delayed, and retrieve again if original history changes. Never reuse returned text for another turn. Shadow, off, and unknown/expired mode return empty immediately; see runtime modes. For synchronous cancellation, pass a threading.Event as cancel. A set event produces delivery == "canceled"; preserve your application’s cancellation check before calling the model. Async callers can use await mirrors.aget_context(trace, timeout_ms=1000, cancel=event) without blocking their event loop. Async task cancellation propagates as asyncio.CancelledError.

Trace capture

capture_trace(trace, final=False, terminal=None) returns a DeliveryTicket without waiting for network delivery. trace is a dictionary of the full original history, excluding Mirrors-added context. Canonical field names stay camelCase, including conversationId, conversationRevision, entryPoint, and tool callId. See trace capture for message structure and conversation IDs. After the model/tool loop and response stream end, capture the full final history with final=True and terminal="completed", "failed", or "canceled". Terminal status describes the execution ending, not task success. Receipt statuses are pending, accepted, dropped, and unconfirmed. When receipt.ingestion is present, receipt.ingestion["status"] == "queued" means analysis was queued, not completed. unconfirmed may follow a committed write; it is not proof of rejection. Ticket waits accept 0–30 seconds; other SDK timeout parameters use milliseconds. Snapshots are limited to 1,000,000 encoded bytes and 10,000 messages. Use builtin JSON data, without custom objects, cycles or silent truncation.

Lifecycle

FlushResult has accepted, dropped, unconfirmed, and pending counts. In Python, .complete is true only when all covered tickets were accepted (or none were pending at entry). Already-settled tickets and later captures are excluded. Keep individual receipts when their outcomes matter. Background threads do not keep the process alive, and in-memory capture queues cannot survive a crash. Use fresh execution IDs after finalization, release or process restart.

Optional user and turn timing

Confirm service compatibility before adding these canonical fields:
  • user: {"id": "internal-user-id", "segment": "cohort"}. Use an opaque internal ID, not an email; segment is optional. The final snapshot’s value is recorded. Trimmed limits are 1–256 characters for ID and 1–64 for segment, without control characters.
  • turnLatencyMs: whole-turn duration on final captures only. Start turn = mirrors.start_turn() when the application accepts the message, before retrieval; use turn.finish() after all model calls, tools, retries and streaming finish. The first finish freezes milliseconds. An existing application measurement can supply a finite nonnegative value; omit unknown duration.
Neither optional field changes model execution. Never send turnLatencyMs with get_context or a non-final capture.