Rust SDK

Install

[dependencies]
introspection-sdk = "0.13"
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.

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.

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

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.

Read what happened

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. 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:

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.

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.

Record feedback

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

introspection-sdk = { version = "0.13", 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.

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