Python SDK

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.

Run a task

client.runtimes(name).run() opens a runner: one session against a deployed runtime. Every argument is optional, 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.

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,
    timeout=300.0,
):
    render(event)

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

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.

Read what happened

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, and 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",
})

Record feedback

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. IntrospectionLogs also provides flush, set_user_id, set_anonymous_id, set_agent, and set_baggage.

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