introspection-sdk 0.19.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 INTROSPECTION_EGRESS_URL, INTROSPECTION_ENDPOINT_HOSTS, and standard proxy environment contract. Transport types cannot be shared between the two HTTP libraries, so only their thin library-specific wrappers differ.

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

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.

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, and metrics.

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.

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.

Environment variables

export INTROSPECTION_TOKEN="intro_xxx"
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