introspection-sdk 0.31.0

The infrastructure for long-horizon vertical agents.

Introspection is the infrastructure for long-horizon vertical agents, powered by Pi. Define an agent as a Recipe — agents, skills, policies, and evals in plain source you own in Git — deploy it to a governed per-customer Runtime, and improve it in production with conversations, observations, judges, and experiments.

This is the Python SDK: run tasks against a deployed runtime, stream their output, and record what users thought of the result.

Install

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

Endpoint-binding proxy transports

The default install includes the native httpx2 adapter:

from introspection_sdk.proxy.httpx2 import IntrospectionTransport

Libraries that still use legacy httpx, including current Harbor and E2B releases, can opt into that dependency and import the matching adapter:

pip install "introspection-sdk[proxy-httpx]"

from introspection_sdk.proxy.httpx import IntrospectionTransport

Both imports use the same routing implementation and the standard proxy environment contract: HTTPS_PROXY/HTTP_PROXY (credentials in the proxy URL are sent as Proxy-Authorization), NO_PROXY, and SSL_CERT_FILE, which also verifies an https:// proxy. Transport types cannot be shared between the two HTTP libraries, so only their thin library-specific wrappers differ.

Runtimes that predate INTROSPECTION_EGRESS_PROXY_NATIVE=1 still set INTROSPECTION_EGRESS_URL and INTROSPECTION_ENDPOINT_HOSTS; requests to those hosts are rewritten to the egress. With the native flag set, or without an egress URL, nothing is rewritten. The rewriting is deprecated and will be removed.

Run a task

import asyncio

from introspection_sdk import AsyncIntrospectionClient


async def main() -> None:
    async with AsyncIntrospectionClient() as client:  # token from INTROSPECTION_TOKEN
        runner = await client.runtimes("customer-agent").run()

        async with runner:
            run = await runner.tasks.start(prompt="Say hello in one sentence.")

            async for event in run.stream():
                print(event)


asyncio.run(main())

runtimes("customer-agent").run() is one request, POST /v1/runtimes/customer-agent/run: the Control Plane resolves the slug in the credential's project (or the project= you pass), so a credential that may not list runtimes, such as a customer signed in by email code, can still open a runner. runner.refresh() posts the same path. A UUID selector is a runtime group id; it is resolved with GET /v1/runtimes first, and client.runtimes.handle(runtime_id) posts a known runtime id directly.

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

IntrospectionClient is the synchronous twin with the same surface — drop the awaits and use for instead of async for.

See Tasks and streaming for reconnects, interrupts, and cancellation.

Record feedback

Install the OpenTelemetry extra, then attach the outcome to the conversation the agent produced:

pip install 'introspection-sdk[otel]'

from introspection_sdk import IntrospectionLogs

logs = IntrospectionLogs(service_name="support-api")

with logs.identify("user_123", traits={"plan": "pro"}):
    with logs.set_conversation(conversation_id):
        logs.feedback("thumbs_up", comments="The answer solved it")

logs.track("case_closed", {"source": "web"})
logs.shutdown()

feedback records how a result landed, track records a product event, and identify attaches who it was.

To record an app event under your own name (ark.feed.entry), use logs.log_event(name, attributes, event_id=...); track is an alias of it. See Logging custom events for idempotency, reserved names, use from a recipe sandbox, and reading events back.

See Product signals for the full surface, and docs/otel.md for the OTel wiring.

Read what happened

A finished task leaves a durable conversation. Add immutable, filter-only metadata when creating the task, then use the same keys to find it later:

await runner.tasks.create(
    prompt="Handle this checkout",
    conversation_metadata={"flow": "checkout", "tenant": "acme"},
)

async for summary in runner.conversations.list(
    limit=20,
    metadata={"flow": "checkout"},
):
    print(summary.id, summary.usage.total_tokens, summary.cost.usd)

The runner also exposes files, shares, events, metrics, automations and connections.

Export a conversation as a Harbor trajectory

spans_to_atif projects the GenAI spans of a conversation onto one ATIF trajectory, the format Harbor reads. It is a pure function of the spans and needs no harbor install:

from pathlib import Path

from introspection_sdk.atif import spans_to_atif

spans = client.conversations.export_json(conversation_id).data
trajectory = spans_to_atif(spans, session_id=conversation_id)
Path("trajectory.json").write_text(trajectory.model_dump_json(indent=2))

Each model call becomes one agent step with its text, reasoning, tool calls, usage and cost, and each tool result is attached to the step that made the call. A total in final_metrics is set only when every model call reported it. For a streamed read, AtifProjection takes one span at a time and returns the steps that are final. Runnable: conversation_atif.py.

One data-plane interface on the client and the Runner

IntrospectionClient and Runner expose the same data-plane namespaces: tasks (with tasks.runs), files, conversations, events, metrics, shares, automations and connections. The contract is introspection_sdk.protocols.DataPlaneResources (and AsyncDataPlaneResources for AsyncIntrospectionClient / AsyncRunner), so code written against it runs with either handle:

from introspection_sdk.protocols import DataPlaneResources


def connected_apps(dp: DataPlaneResources) -> list[str]:
    return [connection.app for connection in dp.connections.list()]


connected_apps(client)                               # the client's own token
connected_apps(client.runtimes("support-agent").run())  # the runner's token

The two differ only in the credential they send: the client sends its own token (an API key, a service account or a member's token) against dp_url, and a Runner sends the session token minted for it. A route the credential lacks the scope for answers 403 either way. A runner a member opens for themself is being given automations:read and automations:write, so it manages that member's automations.

Curate traces with human review

Annotations are append-only events on an OTel trace/span. Each write changes exactly one dimension; label and reviewer lists are complete snapshots, so an empty list clears that dimension.

from introspection_sdk import IntrospectionClient

client = IntrospectionClient(
    token=member_access_token,
    cp_session=encoded_member_session,
    base_api_url="https://api.introspection.dev",
    dp_url="https://dp.example",
)

client.annotations.create(
    trace_id="0af7651916cd43dd8448eb211c80319c",
    span_id="b7ad6b7169203331",
    reviewer_emails=["expert@example.com"],
)
client.annotations.create(
    trace_id="0af7651916cd43dd8448eb211c80319c",
    span_id="b7ad6b7169203331",
    comment="The answer missed the governing exception.",
)

for item in client.annotations.list(label="needs-review"):
    print(item.trace_id, item.span_id, item.latest_comment)

Reusable labels live in client.project_labels; their slug and color are immutable after creation, while the optional description can be updated.

Browse repositories

client.repositories lists the Git repositories linked to a project and reads their contents at one resolved commit through the data plane:

repo = client.repositories.list(slug="acme/support")[0]

for entry in client.repositories.contents(repo.id, "agents", ref="main"):
    print(entry.type, entry.path)

readme = client.repositories.contents.get(repo.id, "README.md")
print(readme.type, readme.commit_sha)

Iterating contents() follows the page cursor; contents.get() returns a RepositoryDirectory page or a RepositoryFile, discriminated on type.

History pages the same way, and one commit carries its changed files and diff:

for commit in client.repositories.commits(repo.id, sha="main", path="agents"):
    print(commit.sha[:7], commit.message.splitlines()[0])

detail = client.repositories.commit(repo.id, commit.sha)
print([f.filename for f in detail.files], len(detail.patch))

Merging mirrors GitHub's merges API: head is a branch or a full commit sha, and None means base already contains it. A conflict raises ConflictError with nothing changed.

merge = client.repositories.merges.create(repo.id, "main", "feature")
if merge is not None:
    print(merge.sha, merge.parents)

Label members

client.members reads and updates the organization's members. Each carries two label sets: tags are access-bearing (a member is admitted by every share whose granted_tag it holds, so writing them needs members:manage), while metadata is a key: value map that grants nothing and exists to filter on. The older implicit grant, where a member reaches every file and task whose tags intersect its own, still works but is being retired; share with a cohort through shares.create(..., granted_tag="team:acme") instead.

for member in client.members.list(metadata={"plan": "enterprise"}, tag="team:acme"):
    print(member.id, member.metadata)

member = client.members.update(member_id, metadata={"plan": "pro", "region": "eu"})
client.members.update(member_id, metadata={})  # clear it

Share files, issues and conversations

shares grants access to a resource you own. The grantee fields are ANDed: granted_member_id admits one member, granted_tag admits everyone whose token carries that tag (you must hold it), both admit that member only while they hold the tag, and neither shares with the whole project. A share admits; the caller's token scopes decide whether they may read, write or delete. Shares are ambient, so a shared resource shows up in the grantee's ordinary reads.

share = client.shares.create(
    resource_type="file", resource_id=file_id, granted_tag="team:acme"
)

# On a conversation, `visible_from` hides earlier spans; `update` changes only it.
convo = client.shares.create(
    resource_type="conversation", resource_id=conversation_id, visible_from=cutoff
)
client.shares.update(str(convo.id), visible_from=None)  # clear the cutoff

update replaces tags and metadata wholesale: omit a field to leave it, pass [] / {} to clear it. The metadata filter ANDs up to 16 pairs with distinct keys, each matched exactly. Keys are letters, digits, _ and - (no .); values are non-empty strings; one write carries at most 64 entries. The server answers 422 otherwise.

create invites a human by email and can seed both sets. A customer member is minted from an asserted identity instead, and that identity can label it:

runner = client.runtimes("customer-agent").run(
    identity={"user_id": "u_123", "metadata": {"plan": "enterprise"}},
)

Identity metadata seeds a new member and is merged into an existing one, overwriting keys of the same name (including ones an admin set) and keeping the rest; refresh() re-sends it. Identity tags, by contrast, apply only when the member is created.

See Production evidence for transcripts, typed events, and metrics queries, Files and shares for durable inputs and grants, and examples/ for end-to-end scripts.

Authenticate

An API key (INTROSPECTION_TOKEN) is the simplest credential. To issue tokens yourself, register an Application in the project. Its type is chosen at creation, cannot change, and gives it exactly one way in; an app that needs two ways in registers two Applications.

The server derives an Application's grants from its type; a client never sends them. An spa needs at least one redirect_uris entry and a jwks takes none; an spa with a brokered identity provider also exchanges that login's id_token through token_exchange. Applications created before the types became exclusive keep working as they did. Tokens for end users belong to a customer member and carry at most the Application's allowed_scopes. A service_account or jwks token is not refreshable: mint or exchange again before expires_in lapses.

Native email-code sign-in

from introspection_sdk import EmailCodeAuth

auth = EmailCodeAuth(client_id="intro_app_...", project="ark")
auth.send_code("user@example.com")
auth.verify_code("user@example.com", code)  # six digits, or six letters and digits on a first sign-in

client = auth.client()  # Data Plane URL from the session
for event in client.events.list("introspection.feedback", limit=5):
    print(event.payload.name)

EmailCodeAuth refreshes the access token leeway seconds (default 60) before it expires and after a 401, and concurrent callers share one refresh. A refresh the server rejects signs the user out and raises AuthenticationError. A sign-in or refresh response that arrives after a newer sign-in or a sign-out is dropped and raises SignInSupersededError, so it never overwrites the newer session. Pass session= to restore a saved AuthSession and on_session_change= to persist each change (None after sign_out()); the session holds the refresh token, so store it as a secret. A rate-limited send_code raises RateLimitError with retry_after.

The token is a Data Plane credential: Data Plane namespaces such as client.events accept it within the Application's allowed_scopes, and Control Plane namespaces such as client.runtimes reject it. AsyncEmailCodeAuth is the asyncio twin, and send_email_code, email_code_token, refresh_access_token and revoke_session are the one-shot calls underneath. See examples/introspection_examples/api/native_email_code.py.

Schedule automations

client.automations manages a project's automations on the data plane: a prompt on a schedule (kind=None), or platform work (observation_synthesis, observation_clustering, project_check_in). A prompt automation creates a task per firing, or posts into an existing task when it names task_id. A one-off reminder is a manual automation with a future next_trigger_at:

from datetime import UTC, datetime

from introspection_sdk.schemas.automations import AutomationMetadata

reminder = client.automations.create(
    name="Friday check-in",
    trigger_type="manual",
    prompt="How did the week go?",
    runtime_group_id=runtime_group_id,
    task_id=task_id,
    next_trigger_at=datetime(2026, 10, 9, 16, tzinfo=UTC),
)

weekly = client.automations.create(
    name="Weekly digest",
    trigger_type="cron",
    cron_schedule="0 9 * * 1",
    prompt="Summarize my week",
    runtime_group_id=runtime_group_id,
    metadata=AutomationMetadata(timezone="Europe/London"),
)

for automation in client.automations.list(enabled=True, scheduled=True):
    print(automation.name, automation.kind, automation.next_trigger_at)

client.automations.update(weekly.id, enabled=False)  # pause, keep the slot
result = client.automations.trigger(weekly.id)       # run it now
print(result.status, result.task_id, result.reason)

update sends only the fields you pass, so None never clears anything; metadata replaces wholesale, and kind / trigger_type are immutable. delete soft-deletes, except for a project default, which answers 409: disable it instead. kind, trigger_type, condition types, skip reasons and trigger statuses are open enums: a value this SDK does not know yet decodes as a plain string rather than failing.

Each firing is recorded as an event, read like any other family:

for event in client.events.list(
    "introspection.automation.triggered", automation_id=weekly.id
):
    print(event.payload.slot, event.payload.task_id, event.payload.posted)

introspection.automation.skipped records a scheduled slot that ran nothing, with its reason; it is project-owned, so only callers with project-wide telemetry access read it. Both families take the automation_id and task_id filters.

Today the server serves /v1/automations to project administrators only and answers anyone else with a 403. introspection-cloud#3137 (not yet shipped) opens the routes to members for their own automations that post into one of their own tasks, and adds the task_id list filter, which this SDK already sends.

The routes need the automations:read and automations:write scopes, so a native sign-in cannot use them yet. can_manage says whether the caller may change an automation, and created_by_member_id who created it.

Connect apps for a member

client.connections / runner.connections manage the apps (Gmail, Slack, …) members connected for themselves, which the agent then acts with in that member's sessions. These are not a connector's connections, which are client.connectors.connections on the control plane.

page = runner.connections.create(app="gmail")  # runtime: the runner's group
print(page.authorize_url, page.expires_in)     # open it once; never cache it

for connection in runner.connections.list(app="gmail"):
    print(connection.id, connection.account_name, connection.healthy)

runner.connections.delete(connection.id)

On the client, create takes the runtime explicitly: client.connections.create(app="gmail", runtime="support-agent"), a runtime slug or runtime group id. list takes member_id and app filters; a caller who is not a project administrator only ever gets their own connections. The routes need connections:read, connections:write and connections:delete.

Run an AAuth Person Server or Access Server

introspection-aauth is a separate package in this repository: an AAuth Person Server and Access Server (draft-hardt-oauth-aauth-protocol -11) over a store you provide, with FastAPI routers. examples/introspection_examples/aauth/four_party.py runs the four-party flow in one process.

Environment variables

export INTROSPECTION_TOKEN="intro_xxx"
export INTROSPECTION_BASE_API_URL="https://api.introspection.dev"  # optional
export INTROSPECTION_SERVICE_NAME="my-service"   # optional
export INTROSPECTION_LOG_LEVEL="debug"           # optional

Documentation

• Python quickstart
• Tasks and streaming
• Files and shares
• Production evidence
• Product signals
• Platform operations
• Python SDK reference
• Authentication

License

Apache-2.0