Python SDK

This page is the Python API reference. Start with Run a task, or read Integrate into a full-stack app for the application lifecycle before translating it to Python.

Install

pip install introspection-sdk
# feedback, track, and identify need the OpenTelemetry extra:
pip install 'introspection-sdk[otel]'
# Arrow exports need pyarrow:
pip install 'introspection-sdk[arrow]'

Python 3.11 or later. Set INTROSPECTION_TOKEN before you run anything — see Authentication.

Follow the application lifecycle

For a normal product integration, read these sections in this order:

1. Run a task and retain its task and run IDs.
2. Continue, control, and reconnect without duplicating work.
3. Read the durable conversation after the turn settles.
4. Record feedback against the exact conversation or response.

Files, tags, connectors, and experiments are optional capabilities. Add them only when the application workflow requires them.

Run a task

client.runtimes(name).run() opens a runner: one session against a deployed runtime. Every argument is optional and keyword-only, so a runner with no identity is the default; the example below binds one by passing identity, which is what lets later work join back to that user. tasks.start() creates a task and hands back a handle on its first run.

run() accepts:

• identity — a RunnerIdentity or plain dict with any of user_id, anonymous_id, and conversation_id, plus tags (see Tags);
• caller — a RunCaller observability payload (IP, user agent, locale, library, page), echoed back on the runner’s context;
• agent_name — which recipe agent runs; omitted uses the recipe default;
• ttl_seconds — session lifetime, default 3600, max 24 hours;
• scope — a narrower scope string for the minted session.

The runner exposes six API clients — tasks, files, conversations, events, metrics, and shares — plus runner.context, a read-only RunnerContext holding what the run resolved to: runtime, recipe, experiment arm, identity, and the echoed caller.

import asyncio
 
from introspection_sdk import AsyncIntrospectionClient
from introspection_sdk.schemas.agui import (
    TextMessageChunkEvent,
    TextMessageContentEvent,
)
 
 
async def main() -> None:
    async with AsyncIntrospectionClient() as client:  # token from INTROSPECTION_TOKEN
        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 | TextMessageChunkEvent,
                ):
                    print(event.delta or "", end="")
 
 
asyncio.run(main())

Or wait for the finished answer instead of streaming:

run = await runner.tasks.start(prompt="Summarize my open tickets.")
print(await run.text())

Continue the same task with a follow-up run:

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())

kind="steer" guides work that is already in progress instead of starting a new turn.

IntrospectionClient is the synchronous twin with the same surface: drop the awaits and use for instead of async for. The runner is a context manager on both sides, so with runner: still holds, but the sync client is not — close it with client.shutdown().

Continue and control a task

Reconnect. Save the task and run ids and a different process can attach to a live run without the original handle. The stream re-attaches by itself across a mid-turn disconnect and raises only once recovery is exhausted.

async for event in runner.tasks.runs.stream(
    saved_task_id,
    saved_run_id,
    max_reconnects=5,
    backoff=0.5,
    timeout=300.0,
):
    render(event)

max_reconnects (default 5) bounds consecutive reconnects that deliver nothing, backoff (default 0.5 seconds) is the base step of the capped exponential retry delay, and timeout (default 300.0 seconds) is the wall-clock deadline for the whole turn.

Answer an interrupt. A run can pause for input — an approval, a missing detail. The RUN_FINISHED event carries the pending interrupts in event.outcome.interrupts; take the one you are answering from there (interrupt = event.outcome.interrupts[0]), then answer it with a resume run:

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

Use status="cancelled" when the caller declines. Several interrupts can be answered in one call.

Cancel. Closing a stream does not stop the agent — cancel the run explicitly.

await run.cancel()  # abort now
await run.cancel({"mode": "drain", "drain_within_seconds": 60})

Read what happened

After the turn settles, read the durable conversation associated with the task instead of treating the stream as the record. Preserve the task and conversation IDs so reconnects, exports, and later evidence all point to the same work. The complete read surface is in Conversation, event, and metrics reference.

Record feedback

Record the user’s outcome against that exact conversation or response. This is what joins product behavior back to the agent work that produced it. Use Feedback API reference for sync and async examples.

Optional capabilities

Skip this group on a first integration. Add only the capabilities your product needs, then continue to Read what happened.

Warming an interactive task

An interactive application can hide sandbox startup behind the time a user spends composing their first message:

1. When the user shows intent to chat, establish the session and create a task without a prompt. Creating it is what starts the sandbox, so the warming has already begun here.
2. When the user submits, adopt that task, update its placeholder title or metadata, and open the first prompt run.
3. If the draft is abandoned, archive the unbound task. Track unfinished cleanup locally so a later page load can retry it.

A bodyless run request is only needed to restart a task that has already finished. On a task created this way it is a no-op that reports status.

# Begin while the user is composing. Creating the task provisions its sandbox,
# so the wait overlaps with typing. `warmup` is application metadata here, not
# a platform switch.
runner = await client.runtimes("support-agent").run(
    identity={"user_id": application_user.id},
)
 
created = await runner.tasks.create(
    agent_name="agent",
    title="Preparing chat",
    idle_timeout_seconds=600,
    metadata={"warmup": True},
)
task_id = str(created.task.id)
 
# Adopt the warm task when the user sends their first message.
await runner.tasks.update(
    task_id,
    title="Research the account",
    metadata={"warmup": False},
)
run = await runner.tasks.runs.create(
    task_id,
    kind="prompt",
    prompt={"text": user_message},
)

Warm only a fresh, unbound draft. An already selected task has its own lifecycle; do not provision it merely because its composer receives focus. Coalesce repeated draft signals and impose a freshness window so typing, focus, and navigation do not create duplicate tasks. Warmup is an optional latency optimization, not a separate task type — keep the normal path as a fallback: if creation or provisioning fails, create the task when the user submits and start the prompt normally.

Resumable streams

A run’s live stream can be severed (a network blip, a gateway idle timeout) without losing anything. stream() re-attaches for you with the last event id it saw (the SSE Last-Event-ID header) and the run replays its recent frames; anything older than the replay window can be hydrated from the task’s conversation record, deduplicating by item id.

A run that is not attachable yet answers the attach with 429 and a Retry-After header. The SDK treats that as a readiness wait, not a failure: it honours Retry-After as the backoff floor and retries, bounded by timeout rather than max_reconnects — a run that is slow to provision has not failed at anything. Only when the deadline lapses does the RateLimitError surface. (At the REST level, wait_for_start=0 on the stream request is the poll-instead-of-hold variant behind this contract.)

Files and shares

Upload a file, then name its id when you start the task so the runtime mounts it before the turn begins:

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

files also has create_text, list, get, update, delete, download, download_stream (an iterator of byte chunks), and the nested files.versions client.

A share is a read-only grant on one conversation or file. The conversation id comes off the conversations list (or off any span you already hold). Pass granted_member_id to target one member, or omit it for a project-wide read grant — an end customer is a member, so there is no separate identity target:

page = await runner.conversations.list(limit=1)
conversation_id = page.records[0].id
 
share = await runner.shares.create(
    resource_type="conversation",
    resource_id=conversation_id,
    granted_member_id=member_id,
)
print(share.url)

Pass a conversation share id as fork_share_id to tasks.create() to branch a new task off that shared history.

Tags

Tags group tasks and, because access is set intersection, share them with a cohort of members. Set them on create, replace them on update, filter with them on list, and read them back off the task.

created = await runner.tasks.create(
    prompt="Draft the Q3 summary.",
    tags=["customer:acme", "team:research"],
)
 
async for task in runner.tasks.list(tag="customer:acme"):
    print(task.id, task.tags)
 
await runner.tasks.update(str(created.task.id), tags=[])

tags on update replaces the list wholesale — unlike metadata, which merges. Omit it to leave tags alone; pass [] to clear them.

Files take the same three operations — tags on files.update, tag on files.list, and tags on the file itself:

await runner.files.update(file_id, tags=["customer:acme"])
 
async for file in runner.files.list(tag="customer:acme"):
    print(file.name, file.tags)

Tags belong to the file rather than to a version, so they carry forward when a new version is written.

tasks.start() accepts everything tasks.create() does, tags included — it is the same call plus a streaming handle.

Stamp a cohort onto an end user as they first appear with identity tags:

runner = await client.runtimes("support-agent").run(
    identity={"user_id": "user_123", "tags": ["customer:acme"]},
)

Those tags apply only when the assertion mints a new customer member, and are narrowed server-side to tags the asserting agent member already holds. See Tags for the access rule and its bounds.

Connectors

client.connectors is a Control Plane namespace, so it hangs off the client rather than a runner, and acts on the project carried by the authenticated credential. It exists for the B2B2C flow: define a provider integration once, then hand each of your own customers an install link so their account connects to an agent. See Connectors for the model.

connector = client.connectors.create(
    name="Slack (support)",
    slug="slack-support",
    provider="slack",
    auth_mode="oauth_stored",
    scopes=["chat:write", "channels:read", "app_mentions:read"],
    api_hosts=["slack.com"],
    client_id=os.environ["SLACK_CLIENT_ID"],
    client_secret=os.environ["SLACK_CLIENT_SECRET"],
)
 
install = client.connectors.authorize(
    connector.id,
    runtime="support-agent",
    expires_in=3600,
)
print(install.authorize_url, install.expires_at)

Create is idempotent on slug, so re-running returns the existing connector. client_secret and signing_secret are write-only — accepted on create and update, absent from every read — and omitting one on update leaves the stored secret unchanged rather than clearing it.

authorize mints the URL you put in front of a customer. It embeds a single-use state, so never cache the result: two calls give two different URLs. Read connector.requires_runtime rather than hardcoding which providers are chat providers — when it is True, omitting runtime is a 422. Pass identity to attribute the grant to your own end user; that mints a customer member, so it can raise ConflictError (409) when the org has reached its member limit.

connections is nested under connectors and every method takes the connector id first, mirroring the routes:

for connection in client.connectors.connections.list(connector.id):
    print(connection.id, connection.subject_type, connection.status)
 
client.connectors.connections.revoke(connector.id, connection_id)

connections also has get, create (registered mode, when you already hold a provider token), and get_token, which resolves a connector credential for the authenticated subject — for a person_authorized connector it can return authorization_pending with a mission and approval URL instead of a token. connectors itself has list, create, get, update, delete, and authorize. AsyncConnectors is the async twin with the same surface.

Conversation, event, and metrics reference

A finished task leaves a durable conversation. Every list() returns a pager: iterate it to walk every page lazily, or take just the first page — await the pager on the async client, pager.page() on the sync one.

async for conversation in runner.conversations.list(limit=20, order="desc"):
    print(conversation.id, conversation.usage.total_tokens, conversation.cost.usd)

Items are the individual turns, always newest-first:

async for item in runner.conversations.items.list(conversation_id):
    gen_ai = item.attributes.gen_ai
    if gen_ai and gen_ai.output:
        render(gen_ai.output.messages)

An item carries the turn-local delta: its input_messages are only the messages new to that turn, so reach for items.get(conversation_id, item_id) when you need the full input history behind a span.

To take a whole conversation at once, use export_json, export_trajectory, export_arrow (the arrow extra), or export_stream(conversation_id, "arrow") for raw bytes — the format is positional and takes "json", "trajectory", or "arrow".

runner.events.list(...) reads one typed event family at a time — introspection.feedback, introspection.judgement, introspection.observation, introspection.pattern, introspection.pattern.assignment, introspection.observation_clustering.run:

async for event in runner.events.list(
    "introspection.feedback",
    conversation_id=conversation_id,
):
    print(event.timestamp, event.payload.name, event.payload.comments)

Use runner.metrics.query(...) for aggregates instead of raw spans. Every query takes a view, one or more metrics, and a bounded from_timestamp / to_timestamp window — both timestamps are required on the typed request; there is no relative-lookback shorthand here. The wire contract is those absolute timestamps on every SDK; the Rust SDK’s lookback field is a client-side convenience it lowers to the same window before sending. The query returns data rows of dimensions and metrics, with meta.row_count, meta.row_limit, and meta.truncated telling you whether the window hit the limit. It takes a MetricQueryRequest or a plain dict; see the API reference for the fields each view allows.

result = await runner.metrics.query({
    "view": "observations",
    "metrics": [{"measure": "conversation_id", "aggregation": "count_distinct"}],
    "dimensions": [{"field": "lens"}],
    "from_timestamp": "2026-07-01T00:00:00Z",
    "to_timestamp": "2026-07-08T00:00:00Z",
})

Feedback API reference

IntrospectionLogs (the otel extra) connects what your users did to what the agent did. feedback records how a result landed, track records a product event, and identify attaches who it was.

from introspection_sdk import IntrospectionLogs
 
logs = IntrospectionLogs(service_name="support-api")
 
with logs.set_conversation(conversation_id):
    logs.feedback("thumbs_up", comments="The answer solved it")
 
logs.identify("user_123", traits={"plan": "pro"})
logs.track("case_closed", {"source": "web"})
logs.shutdown()

Scoping a signal to the conversation is what lets the platform join the outcome back to the work that produced it. To anchor feedback to one response instead of the whole conversation, pass previous_response_id — either as the second argument to set_conversation(conversation_id, previous_response_id) or directly as logs.feedback(..., previous_response_id=...); it sets the gen_ai.request.previous_response_id baggage entry. IntrospectionLogs also provides flush, set_user_id, set_anonymous_id, set_agent, and set_baggage.

Experiments

client.experiments(experiment_id) returns a handle that opens a runner for the experiment and brackets its lifecycle with start, end, and cancel. Authoring the experiment itself — arms, goals, create and delete — is CLI or dashboard work.

experiment = client.experiments(experiment_id)
 
await experiment.start()
runner = await experiment.run(identity={"user_id": "user_123"})
# ... run tasks through the experiment's sticky arm ...
await experiment.end()  # stops evidence collection; deploys nothing

cancel() abandons the experiment instead of concluding it. runner.context reports the arm the run resolved to via arm_label. On the sync client, drop the awaits.

Environment variables

Errors

Every HTTP failure raises IntrospectionAPIError or one of its subclasses, carrying status_code, code, request_id, body, and retry_after. Match on the subclass you handle, and catch the base class to catch everything.

The subclasses are AuthenticationError, InsufficientScopeError, RunnerExpiredError, NotFoundError, ConflictError, ValidationError, RateLimitError, SandboxUnavailableError, and NetworkError — the last one for transport-level failures, with status_code 0 because no response arrived. InsufficientScopeError adds missing_capability, the capability the token was missing.

from introspection_sdk import RunnerExpiredError
 
try:
    await runner.tasks.get(task_id)
except RunnerExpiredError:
    runner = await client.runtimes("support-agent").run()

Related

• CLI
• Authentication
• Platform concepts