Rust SDK

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

Install

[dependencies]
introspection-sdk = "0.15"
futures = "0.3"                                            # StreamExt
serde_json = "1"                                           # json! payloads
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }

IntrospectionClient::new falls back to INTROSPECTION_TOKEN when the ClientConfig carries no token, and returns IntrospectionError::TokenRequired when nothing resolves. See Authentication if you do not have a token yet.

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.runtime(...) resolves a deployed runtime; .run(...) opens a runner, one session against that runtime, through which every read and write below goes; tasks().start_prompt(...) creates a task and returns a handle on its first run.

RunRequest has five fields, all optional (Default gives an anonymous runner):

• identity: Option<RunnerIdentity> — any of user_id, anonymous_id, and conversation_id, plus the member-stamping tags (see Tags);
• caller: Option<RunCaller> — a segment.io-style observability payload (IP, user agent, locale, library, page), echoed back on the runner’s context;
• agent_name: Option<String> — which recipe agent runs; None uses the recipe default;
• ttl_seconds: Option<u32> — session lifetime override, max 24 hours; the server default is one hour;
• scope: Option<String> — a narrower scope string for the minted session.

runner.context() returns the resolved RunnerContext: runtime, recipe, experiment arm (arm_label), identity, and the echoed caller.

use futures::StreamExt;
use introspection_sdk::{AgUiEvent, ClientConfig, IntrospectionClient, RunRequest};
 
let client = IntrospectionClient::new(ClientConfig::default())?;
let runner = client
    .runtime("support-agent")
    .await?
    .run(RunRequest::default())
    .await?;
 
let mut events = runner
    .tasks()
    .start_prompt("Say hello in one sentence.")
    .await?
    .into_stream()
    .await?;
 
while let Some(event) = events.next().await {
    if let AgUiEvent::TextMessageContent(message) = event? {
        print!("{}", message.delta);
    }
}

Close the session with runner.close() and release the client with client.shutdown()? when you are done.

When only the final answer matters, skip the stream:

let handle = runner.tasks().start_prompt("Summarize the open tickets.").await?;
println!("{}", handle.text().await?);

Continue the same task with a follow-up run, where task_id is handle.run.task_id.to_string(). Use TaskRunKind::Steer instead to guide a turn that is already running.

use introspection_sdk::{TaskPrompt, TaskRunCreate, TaskRunKind};
 
let follow_up = runner.tasks().runs.create(&task_id, &TaskRunCreate {
    kind: Some(TaskRunKind::Prompt),
    prompt: Some(TaskPrompt {
        text: "Now draft the reply.".into(),
        images: None,
    }),
    ..Default::default()
}).await?;
 
println!("{}", follow_up.text().await?);

Continue and control a task

Save the task and run IDs when a user may come back while work is still active, then re-open the stream:

let stream = runner.tasks().runs.stream(&saved_task_id, &saved_run_id).await?;

The stream re-attaches transparently across a mid-turn disconnect. Pass a StreamOptions to stream_with(...) to tune the recovery bounds: max_reconnects (default 5), backoff (default 500ms), timeout (default 300s), and emit_reconnect_events (default false), which surfaces each recovery as an introspection.reconnect CUSTOM event. Unknown future AG-UI event types arrive as AgUiEvent::Unknown rather than ending the stream.

Answer an interrupt with typed resume entries. The RUN_FINISHED event carries the pending interrupts in its outcome, an Option<serde_json::Value>: unwrap it, then read outcome["interrupts"][0]["id"] for the interrupt_id. status is "resolved" when the caller answers and "cancelled" when the caller declines.

use introspection_sdk::{ResumeEntry, TaskRunResume};
 
let resumed = runner.tasks().runs.resume(&task_id, &TaskRunResume {
    resume: vec![ResumeEntry {
        interrupt_id,
        status: "resolved".into(),
        payload: Some(serde_json::json!({ "approved": true })),
    }],
}).await?;

Dropping a stream does not stop the agent. Cancel the run explicitly:

use introspection_sdk::TaskCancelOptions;
 
handle.cancel().await?; // stop now, keep the sandbox warm
 
handle.cancel_with(&TaskCancelOptions::Drain {
    drain_within_seconds: Some(60),
}).await?; // let the turn settle, then tear down

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 the telemetry example.

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.

use std::collections::HashMap;
 
use introspection_sdk::{TaskCreate, TaskPrompt, TaskRunCreate, TaskRunKind, TaskUpdate};
 
// 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.
let created = runner.tasks().create(&TaskCreate {
    agent_name: Some("agent".into()),
    title: Some("Preparing chat".into()),
    idle_timeout_seconds: Some(600),
    metadata: Some(HashMap::from([("warmup".into(), serde_json::json!(true))])),
    ..Default::default()
}).await?;
let task_id = created.task.id.to_string();
 
// Adopt the warm task when the user sends their first message.
runner.tasks().update(&task_id, &TaskUpdate {
    title: Some("Research the account".into()),
    metadata: Some(HashMap::from([("warmup".into(), serde_json::json!(false))])),
    ..Default::default()
}).await?;
let run = runner.tasks().runs.create(&task_id, &TaskRunCreate {
    kind: Some(TaskRunKind::Prompt),
    prompt: Some(TaskPrompt { text: user_message, images: None }),
    ..Default::default()
}).await?;

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. The run 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 stream treats that as a readiness wait, not a failure: Retry-After is honoured as the backoff floor and the attach is retried, bounded by StreamOptions::timeout rather than max_reconnects — a run that is slow to provision has not failed at anything. Opt into emit_reconnect_events to observe each reconnect or readiness wait as an introspection.reconnect CUSTOM event. (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

Bytes go through files() first; a task carries only the reference. FileUpload::from_bytes(bytes, "name.txt") covers in-memory content, files().create_text(&FileCreateText { name, content, mime_type }) writes a text file without a multipart upload, and files().download(...) / download_stream(...) read it back.

use introspection_sdk::{FileType, FileUpload, TaskCreate, TaskFileRef};
 
let uploaded = runner.files().upload(
    FileUpload::from_path("notes.txt").with_file_type(FileType::Upload),
).await?;
 
let handle = runner.tasks().start(&TaskCreate {
    prompt: Some("Summarize the attached notes.".into()),
    files: Some(vec![TaskFileRef {
        id: uploaded.id.to_string(),
        ..Default::default()
    }]),
    ..Default::default()
}).await?;

runner.shares() grants read access to one file or conversation without handing out a project credential. Revoke a grant with shares().delete(...), and pass a conversation grant id as fork_share_id on TaskCreate to branch a new task off that shared history.

use introspection_sdk::{ShareCreate, ShareResourceType};
 
let share = runner.shares().create(&ShareCreate {
    resource_type: ShareResourceType::Conversation,
    resource_id: conversation_id.clone(),
    granted_member_id: Some(member_id),
}).await?;

Tags

Tags group tasks and, because access is set intersection, share them with a cohort of members. TaskCreate and TaskUpdate carry tags, TaskListParams carries tag, and Task reads them back.

use introspection_sdk::{TaskCreate, TaskListParams, TaskUpdate};
 
let created = runner.tasks().create(&TaskCreate {
    prompt: Some("Draft the Q3 summary.".into()),
    tags: Some(vec!["customer:acme".into(), "team:research".into()]),
    ..Default::default()
}).await?;
 
let mut pages = runner.tasks().list(&TaskListParams {
    tag: Some("customer:acme".into()),
    ..Default::default()
});
 
// Clearing needs an explicit empty vec; `None` leaves the tags untouched.
runner.tasks().update(&created.task.id.to_string(), &TaskUpdate {
    tags: Some(vec![]),
    ..Default::default()
}).await?;

tags on update replaces the list wholesale — unlike metadata, which merges. tasks().start(&TaskCreate { .. }) takes the same body as create, so tags reach it too; only the start_prompt(...) shorthand is prompt-only.

Files carry the same three: FileUpdate.tags, FileListParams.tag, and File.tags. Tags belong to the file rather than to a version, so they carry forward when a new version is written.

RunnerIdentity also carries tags, which stamp the customer member an asserted identity mints — applied on member creation only, and 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.

use introspection_sdk::{ConnectorAuthMode, ConnectorAuthorizeParams, ConnectorCreateParams};
 
let connector = client.connectors().create(&ConnectorCreateParams {
    slug: Some("slack-support".into()),
    scopes: Some(vec!["chat:write".into(), "channels:read".into()]),
    api_hosts: Some(vec!["slack.com".into()]),
    client_id: Some(std::env::var("SLACK_CLIENT_ID")?),
    client_secret: Some(std::env::var("SLACK_CLIENT_SECRET")?),
    ..ConnectorCreateParams::new("Slack (support)", "slack", ConnectorAuthMode::OauthStored)
}).await?;
 
let install = client.connectors().authorize(connector.id, &ConnectorAuthorizeParams {
    runtime: Some("support-agent".into()),
    expires_in: Some(3600),
    ..Default::default()
}).await?;
println!("{} (until {})", install.authorize_url, install.expires_at);

Create is idempotent on slug. 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.

authorize mints the URL you put in front of a customer. It embeds a single-use state, so never cache the result. Read connector.requires_runtime rather than hardcoding which providers are chat providers: when it is true, omitting runtime is a 422. Set identity to attribute the grant to your own end user; that mints a customer member, so it can fail with 409 when the org has reached its member limit.

connections is nested on the namespace and every method takes the connector id first:

let mut listing = client
    .connectors()
    .connections
    .list(connector.id, &PaginationParams::default());
 
while let Some(connection) = listing.next().await {
    let connection = connection?;
    println!("{} {}", connection.id, connection.status.as_str());
}
 
client.connectors().connections.revoke(connector.id, connection_id).await?;

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 an authorization-pending result with a mission and approval URL instead of a token. connectors() itself has list, create, get, update, delete, and authorize.

Conversation, event, and metrics reference

A finished task leaves a durable conversation. runner.conversations() reads spans and transcripts, runner.events() reads one typed event family at a time, and runner.metrics() runs bounded aggregate queries.

use introspection_sdk::ConversationListParams;
 
let mut pages = runner.conversations().list(&ConversationListParams::default())?;
 
while let Some(page) = pages.next_page().await? {
    for summary in &page.records {
        println!("{} {} tokens", summary.id, summary.usage.total_tokens);
    }
}

conversations().items.list(...) pages the transcript itself. A listed item carries the turn-local delta: its input messages are only the messages new to that turn, so reach for conversations().items.get(...), which returns the full input history for that span, when you need the complete context behind it. The whole-conversation exports sit on the namespace directly, not on items: export_json(...), export_trajectory(...), export_stream(...) for raw bytes, and export_arrow(...) under the arrow feature.

runner.events().list(&EventListParams::new(IntrospectionEventName::Feedback))? reads one family; the six are feedback, judgement, observation, pattern, pattern assignment, and observation-clustering run.

runner.metrics().query(...) runs one bounded aggregation instead. MetricsQuery takes a view, one or more metrics, and a window given either as start / end or as a relative lookback. The wire contract is absolute from_timestamp / to_timestamp values — lookback is a client-side convenience this SDK lowers to that window before sending, which is why the JavaScript and Python request types carry only the two timestamps:

use introspection_sdk::{Dimension, MetricSpec, MetricsQuery};
 
let result = runner.metrics().query(&MetricsQuery {
    view: "observations".into(),
    metrics: vec![MetricSpec { measure: "conversation_id".into(), aggregation: "count_distinct".into() }],
    dimensions: Some(vec![Dimension { field: "lens".into() }]),
    lookback: Some("7d".into()),
    ..Default::default()
}).await?;

Rows follow the requested view, so they arrive as serde_json::Value — see the REST API reference for the allowed views and measures. The response meta rides the same way: read meta["row_count"], meta["row_limit"], and meta["truncated"] to tell whether the result hit the row limit.

Paginator<T> gives you next_page() for the page envelope, collect_all(max_pages) for a bounded slurp, and automatic pagination as a futures::Stream.

Feedback API reference

Enable the otel feature, then attach the outcome to the conversation the agent produced.

introspection-sdk = { version = "0.15", features = ["otel"] }

use introspection_sdk::otel::{FeedbackOptions, IntrospectionLogs, TrackOptions};
 
let logs = IntrospectionLogs::builder()
    .service_name("support-api")
    .build()?;
 
logs.track("case_closed", Some(TrackOptions::new().with_property("source", "web")));
 
{
    let _user = logs.set_user_id("user_123");
    let _conversation = logs.set_conversation_id(&conversation_id);
 
    logs.feedback(
        "thumbs_up",
        FeedbackOptions::new().with_comments("The answer solved it"),
    );
} // guards clear the context when they drop
 
logs.shutdown()?;

feedback records how a result landed, track records a product event, and identify attaches who it was. Scoping a signal to the conversation is what lets the reads above join the outcome back to the work that produced it. To anchor feedback to one response instead of the whole conversation, set the gen_ai.request.previous_response_id baggage entry — either with a logs.set_previous_response_id(...) guard alongside set_conversation_id, or per event with FeedbackOptions::new().with_previous_response_id(...).

Experiments

client.experiment(experiment_id, project) returns an ExperimentHandle that opens a runner for the experiment and brackets its lifecycle with start, end, and cancel; client.experiments() carries the list and get reads. Authoring the experiment itself — arms, goals, create and delete — is CLI or dashboard work.

let experiment = client.experiment(experiment_id, "my-project");
 
experiment.start().await?;
let runner = experiment.run(RunRequest::default()).await?;
// ... run tasks through the experiment's sticky arm ...
experiment.end().await?; // 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.

Environment variables

Errors

REST calls return introspection_sdk::api::ApiResult<T>, whose error type is IntrospectionAPIError. Match on the variant, and use the status(), code(), request_id(), body(), and retry_after() accessors when you only want a field. Http is the non-2xx case and carries all of them: a machine-readable code (what distinguishes an expired runner JWT from any other 401) and retry_after: Option<Duration>, the server’s own number to wait after a 429 once the transparent retry budget is spent. Transport, Decode, Timeout, Io(std::io::Error) (reading a local file for upload), and InvalidConfig are the non-HTTP variants and carry no status.

The two client surfaces fail differently: IntrospectionClient::new(...) and client.shutdown() return IntrospectionError, while IntrospectionLogs::builder().build()? and logs.shutdown()? return IntrospectionLogsError.

Related

• CLI
• Authentication
• Platform concepts