Python SDK

The package exposes Introspection’s REST API through two clients: AsyncIntrospectionClient (recommended) and the synchronous IntrospectionClient. Two layers sit under each: top-level resources for reading and running runtimes and managing experiments and recipes, and a runner for working inside one execution. The runner is a live handle to a sandbox returned by client.runtimes(name).run().

Install

uv add introspection-sdk
# or
pip install introspection-sdk

The default install is everything you need for the platform API: runtimes, tasks, files, and conversations.

Initialize

AsyncIntrospectionClient is the recommended entry point: everything that touches the network is awaitable, run output streams with async for, and async with gives deterministic teardown. The client reads its token from INTROSPECTION_TOKEN if you don’t pass one. The base API URL defaults to https://api.introspection.dev and can be overridden through base_api_url (or INTROSPECTION_BASE_API_URL); pass additional_headers to attach headers to every request.

import asyncio
 
from introspection_sdk import AsyncIntrospectionClient, EventType
 
 
async def main() -> None:
    async with AsyncIntrospectionClient() as client:  # token from INTROSPECTION_TOKEN
        runner = await client.runtimes("support-agent").run()
 
        async with runner:
            run = await runner.tasks.start(prompt="Say hello in one sentence.")
 
            async for event in run.stream():
                if event.type in (EventType.TEXT_MESSAGE_CONTENT, EventType.TEXT_MESSAGE_CHUNK):
                    print(event.delta, end="")
 
 
asyncio.run(main())

The project is taken from the API key server-side, so most calls need no project_id; pass it per call only to override. See Authentication for how keys are scoped.

Resources at a glance

Runtimes and the runner

client.runtimes("name") returns a handle that resolves the name (or UUID) lazily. Calling .run() on it opens a Runner (or AsyncRunner), a live handle to an execution sandbox.

runner = await client.runtimes("support-agent").run(
    identity={"user_id": "user_123"},
    caller={"ip": "8.8.8.8", "user_agent": "my-app/1.0"},
    agent_name="agent",
    scope="tasks:read tasks:write",
)
 
# runner.dp_endpoint, runner.context, runner.session_id, runner.expires_at
# are available here.
 
await runner.close()

The identity binds the session to an end user. Supply exactly one of:

A runner exposes four namespaces (runner.tasks, runner.files, the read-only runner.conversations, and runner.shares) plus runner.close(), which flips a local flag so later calls fail fast. client.runtimes intentionally exposes only list/get, resolution, and run operations. Create, route, deactivate, or reactivate runtime versions through the CLI or dashboard so application code cannot mutate deployment state by accident.

Runner creation also accepts ttl_seconds. The resolved runner.context records the runtime and runtime group, experiment and arm when applicable, flat recipe revision fields, selected agent_name, end-user identity, and caller metadata for attribution.

See Runtimes for the underlying model.

Tasks and runs

A task is one execution; a run is a segment of work within it. runner.tasks.create(prompt="...") creates a task and returns an object carrying .task and .run. The sugar runner.tasks.start(prompt="...") does the same but returns a RunHandle directly, the handle you stream.

# Create + get a RunHandle in one call.
run = await runner.tasks.start(prompt="Summarize my open tickets")
 
# Stream yields typed AG-UI events.
async for event in run.stream():
    if event.type in (EventType.TEXT_MESSAGE_CONTENT, EventType.TEXT_MESSAGE_CHUNK):
        print(event.delta, end="")

A RunHandle exposes three methods:

active = await runner.tasks.start(prompt="Draft a reply")
 
# Bodyless cancel remains an immediate abort and keeps the sandbox warm.
await active.cancel()
 
# Or, for release/teardown, drain instead of aborting:
# await active.cancel({"mode": "drain", "drain_within_seconds": 60})

To fork from a shared conversation, pass fork_share_id to tasks.create:

run = await runner.tasks.create(prompt="continue", fork_share_id=str(share.id))

Follow-up turns open a new run on the same task via runner.tasks.runs.create(...) with a kind of "prompt", "steer", or "clear":

# A follow-up "prompt" turn. The task id comes from run.run.task_id —
# a RunHandle exposes .task / .run objects, not a flat task_id.
follow_up = runner.tasks.runs.create(
    str(run.run.task_id),
    kind="prompt",
    prompt={"text": "Now draft the reply"},
)

runner.tasks also supports list, get, update, delete, and archive. See Tasks for the lifecycle.

Interrupts and resume

A run can pause to ask the caller for input, for example a tool permission or a confirmation. The task moves to awaiting_user and the stream ends with a RUN_FINISHED event whose outcome is an interrupt carrying one or more Interrupt objects.

from introspection_sdk import EventType, Interrupt, ResumeEntry
 
run = runner.tasks.start(prompt="Refund order #123")
 
interrupts: list[Interrupt] = []
for event in run.stream():
    if event.type == EventType.RUN_FINISHED and event.outcome and event.outcome.type == "interrupt":
        interrupts = event.outcome.interrupts
 
# Answer the interrupts by opening a resume run on the same task.
resumed = runner.tasks.runs.resume(
    str(run.run.task_id),
    resume=[ResumeEntry(interrupt_id=interrupts[0].id, status="resolved", payload={"approved": True})],
)
print(resumed.text())

Each ResumeEntry references an interrupt_id and a status of "resolved" (optionally with a payload) or "cancelled". Resume lives on runner.tasks.runs.resume(task_id, resume=[...]), not on tasks. Resolving every pending interrupt flips the task back to running; answering only some keeps it awaiting_user.

Files

runner.files reads and writes durable files and their versions.

# Create a text file inline.
note = await runner.files.create_text(
    name="summary.md",
    content="# Summary\n...",
    mime_type="text/markdown",
)
 
# Upload raw bytes.
binary = await runner.files.upload(
    file=b"hello binary",
    name="hello.bin",
    file_type="upload",
)
 
# Download raw bytes.
payload = await runner.files.download(str(note.id))

list() returns a Pager (AsyncPager on the async client). Call .page() for the first page with its envelope metadata (.records and .total_count), or iterate it to stream every item across pages, fetched lazily.

# First page with totals.
first = await runner.files.list(include_total=True).page()
print(first.total_count, len(first.records))
 
# Or stream every file across all pages.
async for f in runner.files.list():
    print(f.id, f.name)

Other methods: files.get, files.update, files.delete, and the files.versions sub-namespace for version history.

Conversations

runner.conversations is read-only: the replayable record a task leaves behind.

# List conversation summaries (each page via `.page()`, or stream with `async for`).
convos = await runner.conversations.list(limit=5).page()
for summary in convos.records:
    cid = summary.conversation_id or summary.trace_id
 
    # Responses-API-style snapshot of the latest LLM turn.
    response = await runner.conversations.retrieve(cid)
    if response is not None:
        print(response.model, len(response.input_messages), len(response.output_messages))
 
    # Walk the transcript item by item.
    async for item in runner.conversations.items.list(cid, order="asc"):
        print(item.id, item.node_type)

retrieve() returns None when the conversation has no items.

Every list() returns a Pager (AsyncPager on the async client): call .page() for the first page and its envelope metadata (.records, .total_count, .next), or iterate it (for / async for) to stream every row across pages, fetched lazily. iterate(max_items=N) is a bounded convenience generator over the same stream. A page returns up to 1000 rows (limit, default 100); the SDK follows the opaque next cursor for you.

Events

runner.events is a read-only cursor list over the platform event stream (GET /v1/events), mirroring runner.conversations. Every read names its family: event_name is required, exactly one of the six canonical platform families.

Because a request names one family, every page is homogeneous: each row is that family’s typed model, sharing a common envelope (id, timestamp, event_name, trace_id/span_id, conversation_id, service_name, environment, and lineage — runtime_group_id, runtime_id, experiment_id, recipe_git_commit_sha) plus a nested, family-typed payload. For the two state families the server folds the event history and answers with resolved current state; the four stream families return plain event rows. “A pattern’s observations” is the observation family filtered by pattern:

# Feedback for a conversation, newest first.
async for ev in runner.events.list(
    "introspection.feedback", conversation_id=conversation_id, order="desc"
):
    print(ev.timestamp, ev.payload.name, ev.payload.comments)
 
# A pattern's observations: resolved current state, bounded pull.
observations = [
    e async for e in runner.events.iterate(
        "introspection.observation", pattern_id=pattern_id, max_items=500
    )
]

Envelope filters (time window, conversation_id, service_name, environment, lineage, trace_id/span_id) apply to every family. Family-scoped filters — observation: lens, pattern_id, include_superseded, severities, conversation_ids; pattern: lens, status — are validated server-side against the requested family. Omitting event_name is an error. Rows outside the six families (custom track() events, gen_ai.* records) are not enumerable here; they remain measurable through the metrics API.

runner.metrics runs the bounded aggregate query (POST /v1/metrics) — see Conversations → Querying and measuring.

Efficient bulk downloads (Arrow)

Both runner.conversations.list and runner.events.list can return each page as an Apache Arrow  IPC stream instead of JSON — pass format="arrow" to move the column data over the wire in Arrow’s compact binary layout and skip per-row JSON parsing when pulling large histories:

async for ev in runner.events.list(
    "introspection.judgement", lookback="7d", format="arrow"
):
    print(ev.id, ev.payload.result)

For columnar consumption, skip the row models entirely with the .arrow() accessor: it yields one pyarrow.Table per server page — constant envelope columns plus a typed payload struct column fixed by the requested family (one unnest away from pandas/polars) — and .read_all() concatenates every page into a single Table. It takes the same filters as list(), and runner.conversations.arrow(...) works the same way.

table = await runner.events.arrow(
    "introspection.observation", lookback="30d"
).read_all()
df = table.to_pandas()

Arrow is an optional dependency: the default JSON path never loads it, so install the extra only when you opt in.

pip install "introspection-sdk[arrow]"

With format="arrow" on list(), pagination, filtering, and the returned row models are identical to the JSON path — only the wire encoding changes. Without pyarrow installed, the Arrow paths raise a clear error pointing at the missing extra.

Shares

runner.shares creates read-only grants for a conversation or file. A grant carries a url with the ?share_id capability and an id; pass that id as fork_share_id to fork a task.

share = await runner.shares.create(
    resource_type="conversation",
    resource_id=conversation_id,
    granted_identity_key="user:user_123",
)
print(share.url, share.id)
 
await runner.shares.list()
await runner.shares.get(str(share.id))
await runner.shares.delete(str(share.id))  # revoke

Set either granted_member_id or granted_identity_key to target one recipient; omit both for a project-wide read grant. The two target fields are mutually exclusive.

Experiments

client.experiments(id).run() returns a Runner exactly like a runtime does; drive runner.tasks against it the same way. Use the CLI’s experiments create --file experiment.yaml for the current creation schema, then use the SDK to read, start, run, end, or cancel the experiment. See Experiments.

runner = await client.experiments(experiment_id).run(
    identity={"user_id": "user_123"},
)

Recipes

client.recipes is CRUD over immutable build artefacts (repository + git ref + commit SHA + optional sub-path), referenced by runtimes and experiment arms.

async for recipe in client.recipes.list():
    print(recipe.slug, recipe.git_commit_sha)

Service-account (machine) auth

A long-lived INTROSPECTION_TOKEN is the simplest credential. For headless / CI callers that should not ship a static key, authenticate as a confidential service-account Application: from_service_account mints a short-lived, project-scoped token via the OAuth client_credentials grant and wires it in, so the runtime flow is unchanged.

from introspection_sdk import IntrospectionClient
 
client = IntrospectionClient.from_service_account(
    client_id="intro_app_…",      # confidential Application
    client_secret="intro_sk_…",   # minted once, kept server-side
    project_id="proj_…",          # the token is project-scoped
)
runner = client.runtimes("support-agent").run()

The token is not auto-refreshed; re-mint once it expires. AsyncIntrospectionClient.from_service_account is the awaitable twin. When acting as a server broker handing credentials to a browser client, call service_account_token(...) directly to also read dp_url (the resolved Data Plane endpoint) alongside the access token.

Sync client

Not on asyncio? IntrospectionClient is the synchronous twin with an identical surface: drop the awaits, use for instead of async for, and with instead of async with.

from introspection_sdk import IntrospectionClient, EventType
 
client = IntrospectionClient()  # token from INTROSPECTION_TOKEN
runner = client.runtimes("support-agent").run()
 
run = runner.tasks.start(prompt="Say hello in one sentence.")
for event in run.stream():
    if event.type in (EventType.TEXT_MESSAGE_CONTENT, EventType.TEXT_MESSAGE_CHUNK):
        print(event.delta, end="")
 
runner.close()
client.shutdown()

Related

• Runtimes, Tasks, Experiments: the underlying model.
• Authentication: tokens, scopes, and environments.
• API Reference: the full REST surface.