rpi-agent

The provider-agnostic agent runtime: the agent loop, the AgentTool trait, events, hooks, queues and cancellation. This is the layer you embed when you want to build your own agent rather than use the rpi CLI.

Install

[dependencies]
rpi-agent = "0.3"
rpi-ai = "0.3"
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }

For built-in file and shell tools, add rpi-tools. For durable sessions and compaction, add rpi-harness.

Design

The only LLM boundary is StreamFn, which returns an AssistantMessageEventStream synchronously. Everything else in the loop speaks AgentMessage and never touches a provider wire format, so a provider, a test double and a recorded transcript are interchangeable.

The invariants that the loop guarantees:

• Tools may finish in any order, but tool results are emitted in the order the calls appeared, so the transcript stays valid for the next request.
• A MessageEnd is emitted exactly once per assistant message, including when a late streaming update arrives after the run already settled.
• Truncated or failed tool calls never leave a dangling call without a result.

Quick start

Deterministic, offline, no API key — the faux provider scripts one text reply:

use std::sync::Arc;

use rpi_agent::AgentBuilder;
use rpi_ai::providers::faux::{FauxProvider, FauxScript};
use rpi_ai::Provider;

#[tokio::main]
async fn main() {
    let provider = Arc::new(FauxProvider::new(
        FauxScript::new().with_text("Hello from the faux provider!"),
    ));
    let model = provider.default_model().clone();

    let agent = AgentBuilder::new()
        .model(model)
        .system_prompt("You are a helpful assistant.")
        .stream_fn(make_stream_fn(provider))
        .build()
        .expect("agent builds");

    let mut rx = agent.subscribe();
    agent.prompt("hi").await.expect("prompt completes");

    while let Ok(ev) = rx.try_recv() {
        // AgentEvent::MessageUpdate { .. }, AgentEvent::ToolEnd { .. }, …
        let _ = ev;
    }

    println!("{} messages after the run", agent.state().messages.len());
}

fn make_stream_fn(provider: Arc<FauxProvider>) -> rpi_agent::StreamFn {
    rpi_agent::stream_fn(move |model, ctx, opts| {
        let p = Arc::clone(&provider);
        let (model, ctx, opts) = (model.clone(), ctx.clone(), opts.clone());
        tokio::task::block_in_place(|| {
            tokio::runtime::Handle::current()
                .block_on(async move { p.stream_simple(&model, &ctx, &opts).await })
        })
    })
}

A runnable version of this lives in examples/minimal (cargo run -p minimal), and a version that drives the real filesystem through the built-in tools lives in examples/tools.

Defining a tool

Tools implement AgentTool. A tool owns the provider-facing schema, a display label, and the execution function; parameters are validated against that schema before execute is called:

use std::sync::Arc;

use async_trait::async_trait;
use rpi_agent::{AgentError, AgentTool, AgentToolResult, ToolResultPartial};
use rpi_ai::types::Tool;
use tokio_util::sync::CancellationToken;

struct MyTool {
    schema: Tool,
}

#[async_trait]
impl AgentTool for MyTool {
    fn schema(&self) -> &Tool {
        &self.schema
    }

    fn label(&self) -> &str {
        "My tool"
    }

    async fn execute(
        &self,
        _tool_call_id: &str,
        params: serde_json::Value,
        _signal: CancellationToken,
        on_update: Arc<dyn Fn(ToolResultPartial) + Send + Sync>,
    ) -> Result<AgentToolResult, AgentError> {
        // Long-running tools call `on_update(partial)` to stream progress.
        let _ = on_update;
        Ok(AgentToolResult::default())
    }
}

execute receives a CancellationToken for cancellation and an on_update callback for partial results. Calls to on_update after the run has settled are dropped by the loop, so a tool never has to track settlement itself. Returning Err(AgentError) turns into an error tool result, not a failed run.

For real implementations, wrap the JSON Schema derivation with schemars and start from rpi-tools, which contains complete examples of a filesystem tool and a streaming shell tool.

Observing a run

agent.subscribe() returns a broadcast receiver of AgentEvent covering the run (AgentStart/AgentEnd), each assistant message (MessageStart/MessageUpdate/MessageEnd), and each tool (ToolStart/ToolUpdate/ToolEnd). Cancellation goes through AbortHandle.

Related crates

Documentation

• Crate docs: https://docs.rs/rpi-agent
• Architecture: docs/architecture.md
• Project structure guide: docs/agent-project.md

License

MIT. See LICENSE and NOTICE.