Skip to Content
Platform
ReferencePython SDK

Python SDK

Run tasks against a deployed runtime, stream their output, and record what users thought of the result.

The Python SDK has the same surface as the JavaScript SDK; this page shows the Python calls and what differs.

Install

pip install introspection-sdk
pip install 'introspection-sdk[otel]'   # feedback, track, identify
pip install 'introspection-sdk[arrow]'  # Arrow exports

Python 3.11 or later. The client reads INTROSPECTION_TOKEN; see Authentication.

Run a task

import asyncio
 
from introspection_sdk import AsyncIntrospectionClient
from introspection_sdk.schemas.agui import TextMessageContentEvent
 
 
async def main() -> None:
    async with AsyncIntrospectionClient() as client:
        runner = await client.runtimes("support-agent").run(identity={"user_id": "user_123"})
        async with runner:
            run = await runner.tasks.start(prompt="Summarize my open tickets.")
            async for event in run.stream():
                if isinstance(event, TextMessageContentEvent):
                    print(event.delta or "", end="")
 
            follow_up = await runner.tasks.runs.create(
                str(run.run.task_id), kind="prompt", prompt={"text": "Now draft the reply."}
            )
            print(await follow_up.text())
 
 
asyncio.run(main())
  • run() takes keyword-only identity (user_id, anonymous_id, conversation_id, tags, metadata), agent_name, ttl_seconds, caller, and scope. Who owns the task: Run a task for an end user.
  • IntrospectionClient is the synchronous twin: drop the awaits, use for, and close it with client.shutdown().
  • In Jupyter, await directly in a cell (no asyncio.run), or use IntrospectionClient.
  • kind="steer" redirects a running turn.

tasks.create() takes title, prompt, agent_name, repositories, metadata, conversation_metadata, idle_timeout_seconds, fork_share_id, files, and tags; tasks.start(prompt=..., **kwargs) takes the same. metadata={"model": "<provider>/<model-id>", "thinking_level": "high"} overrides the root agent’s model for one task.

Continue and control a task

Reattach after a reload

task = await runner.tasks.get(task_id)
run_id = task.metadata.get("active_run_id") or task.metadata.get("latest_run_id")
async for event in runner.tasks.runs.stream(task_id, run_id):
    render(event)

There is no since cursor: a reattach replays the run, so render idempotently.

Stream options and errors

stream() takes timeout (default 300.0 s), max_reconnects (5), and backoff (0.5 s), with the same meaning as the JavaScript options: when timeout lapses only your stream raises; the run keeps going. RunFailedError (do not retry) and StreamIncompleteError (the run settled: read the conversation) are raised by the stream and by text(). HTTP calls retry twice (429; 502/503/504 on GET) and time out after 30 seconds per request.

Answer an interrupt

from introspection_sdk import ResumeEntry
 
resumed = await runner.tasks.runs.resume(
    task_id,
    resume=[ResumeEntry(interrupt_id=interrupt.id, status="resolved", payload={"approved": True})],
)

Take interrupt from event.outcome.interrupts on RUN_FINISHED. Payloads and the recipe side: Ask the user.

Cancel

await run.cancel()  # abort the turn, keep the sandbox warm
await run.cancel({"mode": "drain", "drain_within_seconds": 60})

Files

from pathlib import Path
 
uploaded = await runner.files.upload(file=Path("spec.md"), file_type="upload")
run = await runner.tasks.start(prompt="Summarize spec.md.", files=[{"id": str(uploaded.id)}])

files also has create_text, list, get, update, delete, download, download_stream, and versions. list() filters only by name, file_type, storage_path, tag, and metadata; for one task’s outputs use introspection files list --filter. Shares and tags: Sharing.

Conversations, events, and metrics

async for c in runner.conversations.list(limit=20, order="desc"):
    print(c.id, c.cost.usd)
 
# Tool calls only, e.g. a submit_result tool's arguments (a JSON string)
async for item in runner.conversations.items.list(task_id, operation_name="execute_tool"):
    print(item.attributes.gen_ai.tool.name, item.attributes.gen_ai.tool.call.arguments)
 
async for event in runner.events.list("introspection.feedback", conversation_id=task_id):
    print(event.payload.name)
 
result = await runner.metrics.query({
    "view": "conversations",
    "metrics": [{"measure": "total_cost_usd", "aggregation": "sum"}],
    "from_timestamp": "2026-07-01T00:00:00Z",
    "to_timestamp": "2026-07-08T00:00:00Z",
})

Every event family is typed, including AnnotationEvent and TrackEvent. Exports: export_json, export_trajectory, export_arrow (needs [arrow]), export_stream. A list returns a pager: iterate it, or await it for one page (pager.page() on the sync client).

Feedback

from introspection_sdk import IntrospectionLogs
 
logs = IntrospectionLogs(service_name="support-api")
logs.feedback("thumbs_down", conversation_id=task_id, comments="Missed the refund policy", value=0)
logs.track("case_closed", {"source": "web"})
logs.shutdown()

previous_response_id anchors feedback to one response; extra keyword arguments are stored as properties. Name and sentiment rules: Send feedback. IntrospectionLogs keeps one token for its lifetime, so give it an API key.

Experiments

client.experiments has list, get, create(document), update(experiment_id, document), and delete(experiment_id) (sync and async); documents match introspection experiments create. client.experiments(experiment_id) returns a handle with run, start, end, and cancel.

Environment variables

Same as JavaScript, except INTROSPECTION_STRICT_TRACING. INTROSPECTION_BASE_OTEL_URL defaults to https://otel.introspection.dev and INTROSPECTION_SERVICE_NAME to introspection-client.

Errors

Every HTTP failure raises IntrospectionAPIError (status_code, code, request_id, body, retry_after) or a subclass: AuthenticationError, InsufficientScopeError (missing_capability), RunnerExpiredError, NotFoundError, ConflictError, ValidationError, RateLimitError, SandboxUnavailableError, NetworkError, plus RunFailedError and StreamIncompleteError from streams.

Next

Last updated on