Streaming Tools Guide

Aeolus supports long-running, bidirectional tool interactions through the StreamingTool trait. This guide covers how streaming tools work, the four buffering policies, and how to implement custom streaming tools.

Overview

Standard tools (Tool trait) follow a simple request-response pattern: the agent sends input, the tool returns output. Streaming tools extend this with:

Incremental output — results arrive as a stream of chunks, not a single response
Bidirectional communication — the agent can send stdin events to a running tool
Policy-aware buffering — output is buffered, windowed, summarized, or manually controlled before injection into the agent's context
Checkpoint recovery — streaming sessions produce idempotency keys for crash recovery

Agent                          StreamingTool
  |                                |
  |  ctx.stream_tool(tool, input)  |
  |------------------------------->|
  |                                |
  |  StreamingToolSession          |
  |  +-- chunks (Stream)           |
  |  +-- stdin  (Sender)           |  <-- agent can send StdinEvent
  |  +-- cancel (CancellationToken)|
  |  +-- finalize (Future)         |
  |                                |
  |  session.next_chunk()          |
  |<-------------------------------|  OutputChunk
  |  session.next_chunk()          |
  |<-------------------------------|  OutputChunk
  |  session.send_stdin(...)       |
  |------------------------------->|
  |  ...                           |
  |  session.finalize()            |
  |<-------------------------------|  Final

StreamingTool Trait

#[async_trait]
pub trait StreamingTool: Send + Sync + 'static {
    const NAME: &'static str;
    const DESCRIPTION: &'static str;
    type Input:  Serialize + DeserializeOwned + JsonSchema + Send;
    type OutputChunk: Serialize + DeserializeOwned + JsonSchema + Send;
    type Final: Serialize + DeserializeOwned + JsonSchema + Send;

    async fn start(
        &self,
        input: Self::Input,
    ) -> Result<StreamingToolHandle<Self::OutputChunk, Self::Final>, ToolError>;

    fn streaming_policy() -> StreamingToolPolicy { StreamingToolPolicy::Buffer }
    fn isolation() -> IsolationMode { IsolationMode::InProcess }
    fn idempotent() -> bool { false }
    fn required_capability() -> Option<CapabilityKind> { None }
}

Associated Types

Type	Description
`Input`	Configuration for the streaming session
`OutputChunk`	Incremental output items (stdout lines, log entries, etc.)
`Final`	Terminal result (exit code, summary, etc.)

StreamingToolHandle

Returned by start(), the handle provides access to the running session:

pub struct StreamingToolHandle<C, F> {
    pub chunks: Pin<Box<dyn Stream<Item = Result<C, ToolError>> + Send>>,
    pub stdin: Option<mpsc::Sender<StdinEvent>>,
    pub cancel: CancellationToken,
    pub finalize: Pin<Box<dyn Future<Output = Result<F, ToolError>> + Send>>,
}

Field	Type	Description
`chunks`	`Stream<Item = Result<C>>`	Incremental output stream
`stdin`	`Option<Sender<StdinEvent>>`	Channel for sending input to the tool (bidirectional only)
`cancel`	`CancellationToken`	Cancel the running tool
`finalize`	`Future<Output = Result<F>>`	Waits for the tool to complete and returns the final result

StreamingToolSession

The StreamingToolSession wraps the raw handle with policy-aware buffering:

pub struct StreamingToolSession<T: StreamingTool> {
    // internal: handle, policy, chunk buffer, context window tracker
}

Method	Description
`session.next_chunk()`	Get the next chunk (policy may buffer or summarize)
`session.send_stdin(event)`	Send a stdin event to the tool
`session.cancel()`	Cancel the running tool
`session.finalize()`	Wait for completion and get the final result
`session.context_window()`	Get current context window usage

StdinEvent

Bidirectional tools accept stdin events from the agent:

pub enum StdinEvent {
    Text(String),                // Text input
    Bytes(Bytes),                // Binary input
    Signal(UnixSignal),          // Unix signal (SigInt, SigTerm, etc.)
    Resize { cols: u16, rows: u16 },  // Terminal resize
}

pub enum UnixSignal {
    SigInt,    // Ctrl+C
    SigTerm,   // Graceful termination
    SigKill,   // Force kill
    SigHup,    // Hangup
}

StreamingToolPolicy

Four policies control how streaming output is managed:

1. Buffer (default)

Collects all chunks into a single result. Simple and deterministic:

fn streaming_policy() -> StreamingToolPolicy {
    StreamingToolPolicy::Buffer
}

All chunks are collected, then the combined output is returned as a single context injection. Best for short-lived tools where the total output fits in context.

2. Window

Keeps a sliding window of the most recent N chunks:

fn streaming_policy() -> StreamingToolPolicy {
    StreamingToolPolicy::Window { max_chunks: 200 }
}

When the agent reads from the session, it receives the last max_chunks chunks. Older chunks are discarded. Best for long-running tools where only recent output matters (log tailing, terminal output).

3. Summarise

Uses an LLM to periodically summarize accumulated output:

fn streaming_policy() -> StreamingToolPolicy {
    StreamingToolPolicy::Summarise(SummariseConfig {
        chunk_threshold: 50,          // summarize every 50 chunks
        max_summary_length: 2000,     // max characters for summary
        model: ModelSpec::anthropic("claude-haiku-4-5"),
    })
}

The policy accumulates chunks until chunk_threshold is reached, then calls the configured LLM to produce a summary. The summary replaces the accumulated chunks in context. On finalize, a final summarization pass occurs if the accumulated output exceeds max_summary_length.

Falls back to truncation if the LLM call fails.

4. Manual

The agent controls when and how output is injected into context, subject to human approval:

fn streaming_policy() -> StreamingToolPolicy {
    StreamingToolPolicy::Manual(ManualConfig {
        approvers: vec!["role:admin".into()],
        preview_length: 500,
        timeout: Duration::from_secs(300),
        timeout_action: ManualTimeoutAction::Deny,
    })
}

When the agent calls session.next_chunk(), the policy:

Buffers the chunk
Presents a preview (up to preview_length chars) to the configured approvers
Waits for approval decision
Returns the chunk on approval, or returns an error on denial/timeout

Idempotency

Streaming tools can opt into checkpoint-based recovery:

fn idempotent() -> bool { true }

When enabled:

start() is treated as an atomic step boundary
The IdempotencyKey (UUID v7) is recorded in the checkpoint
On crash recovery, if a Final value exists for the key, the tool is not re-executed
The stored Final value is returned directly

Using Streaming Tools from an Agent

async fn run(
    &self,
    input: Self::Input,
    ctx: &mut AgentContext<()>,
) -> Result<Self::Output, AgentError> {
    let mut session = ctx.stream_tool(
        TerminalTool,
        TerminalInput {
            command: "tail -f /var/log/app.log".into(),
            timeout_secs: Some(60),
            ..Default::default()
        },
    ).await?;

    while let Some(chunk) = session.next_chunk().await? {
        // Process each log line
        if chunk.text.contains("ERROR") {
            session.send_stdin(StdinEvent::Signal(UnixSignal::SigInt)).await?;
            break;
        }
    }

    let result = session.finalize().await?;
    Ok(/* process result */)
}

Built-in Streaming Tools

Terminal Tool (`tools/terminal`)

A bidirectional streaming tool for shell command execution:

use tools::terminal::TerminalTool;

let session = ctx.stream_tool(
    TerminalTool,
    TerminalInput {
        command: "python3 -i".into(),
        cwd: Some("/workspace".into()),
        env: vec![("PYTHONPATH".into(), "/lib".into())],
        timeout_secs: Some(120),
    },
).await?;

// Send Python code
session.send_stdin(StdinEvent::Text("print('hello')\n".into())).await?;

// Read output
while let Some(chunk) = session.next_chunk().await? {
    println!("[{}] {}", chunk.stream, chunk.text);  // Stdout or Stderr
}

let result: TerminalResult = session.finalize().await?;
println!("Exit code: {}", result.exit_code);

Properties:

Bidirectional (stdin supported)
Policy: Window { max_chunks: 200 }
Isolation: InProcess (subprocess sandboxing is Phase 2)
Validates commands against dangerous patterns (fork bombs, rm -rf /, etc.)

Log Tailer Tool (`tools/log-tailer`)

A read-only streaming tool for following log files:

use tools::log_tailer::LogTailerTool;

let session = ctx.stream_tool(
    LogTailerTool,
    LogTailerInput {
        path: "/var/log/app.log".into(),
        follow: true,
        format: LogFormat::Json,
        filter: Some(LogFilter {
            level: Some(LogLevel::Error),
            pattern: Some(Regex::new("timeout")?),
            ..Default::default()
        }),
        ..Default::default()
    },
).await?;

while let Some(chunk) = session.next_chunk().await? {
    if let Some(level) = &chunk.level {
        println!("[{}] {}", level, chunk.message);
    }
}

let result: LogResult = session.finalize().await?;
println!("Lines read: {}", result.lines_read);

Properties:

Read-only (no stdin)
Policy: Window { max_chunks: 200 }
Isolation: InProcess
Idempotent: true
Multi-format: JSON, Syslog, CLF, Combined, Structured, Auto, Raw
Supports: follow mode, log rotation detection, regex/level filtering, glob patterns

Code Interpreter Tool (`tools/code-interpreter`)

A read-only streaming tool for executing code in multiple languages:

use tools::code_interpreter::CodeInterpreterTool;

let session = ctx.stream_tool(
    CodeInterpreterTool,
    CodeInterpreterInput {
        language: "python".into(),
        code: r#"
import matplotlib.pyplot as plt
plt.plot([1, 2, 3, 4], [1, 4, 9, 16])
plt.savefig('plot.png')
print("Plot saved!")
"#.into(),
        timeout_secs: Some(30),
        ..Default::default()
    },
).await?;

while let Some(chunk) = session.next_chunk().await? {
    match chunk.stream {
        CodeStream::Stdout => println!("OUT: {}", chunk.content),
        CodeStream::Image => println!("IMAGE: {}", chunk.mime_type),
        _ => {}
    }
}

let result: CodeResult = session.finalize().await?;
println!("Success: {}, Time: {}ms", result.success, result.execution_time_ms);

Properties:

Read-only (no stdin)
Policy: Window { max_chunks: 100 }
Isolation: InProcess (subprocess sandboxing is Phase 2)
Languages: Python, JavaScript, Bash, R (WASM planned)
Supports: sessions, file injection, dangerous code analysis, packages

Implementing a Custom Streaming Tool

Read-Only Streaming Tool

use aeolus_tools::prelude::*;

struct SensorMonitor;

#[derive(Serialize, Deserialize, JsonSchema)]
struct SensorInput {
    device_id: String,
    duration_secs: u64,
}

#[derive(Serialize, Deserialize, JsonSchema)]
struct SensorReading {
    timestamp: u64,
    value: f64,
    unit: String,
}

#[derive(Serialize, Deserialize, JsonSchema)]
struct SensorSummary {
    readings_count: u32,
    average: f64,
    min: f64,
    max: f64,
}

#[async_trait]
impl StreamingTool for SensorMonitor {
    const NAME: &'static str = "sensor_monitor";
    const DESCRIPTION: &'static str = "Monitor a sensor device and stream readings";
    type Input = SensorInput;
    type OutputChunk = SensorReading;
    type Final = SensorSummary;

    fn streaming_policy() -> StreamingToolPolicy {
        StreamingToolPolicy::Window { max_chunks: 50 }
    }

    fn idempotent() -> bool { true }

    async fn start(
        &self,
        input: SensorInput,
    ) -> Result<StreamingToolHandle<SensorReading, SensorSummary>, ToolError> {
        let (chunk_tx, chunk_rx) = mpsc::channel(100);
        let (done_tx, done_rx) = oneshot::channel();

        tokio::spawn(async move {
            let mut count = 0u32;
            let mut sum = 0.0;
            let mut min = f64::MAX;
            let mut max = f64::MIN;
            let deadline = tokio::time::Instant::now()
                + Duration::from_secs(input.duration_secs);

            loop {
                if tokio::time::Instant::now() >= deadline {
                    break;
                }
                let reading = read_sensor(&input.device_id).await;
                count += 1;
                sum += reading.value;
                min = min.min(reading.value);
                max = max.max(reading.value);
                if chunk_tx.send(Ok(reading)).await.is_err() {
                    break;
                }
                tokio::time::sleep(Duration::from_secs(1)).await;
            }

            let _ = done_tx.send(Ok(SensorSummary {
                readings_count: count,
                average: if count > 0 { sum / count as f64 } else { 0.0 },
                min,
                max,
            }));
        });

        Ok(StreamingToolHandle {
            chunks: Box::pin(ReceiverStream::new(chunk_rx)),
            stdin: None, // read-only
            cancel: CancellationToken::new(),
            finalize: Box::pin(async move { done_rx.await.map_err(|_| ToolError::Cancelled)? }),
        })
    }
}

Bidirectional Streaming Tool

Set stdin to Some(tx) to accept stdin events:

async fn start(
    &self,
    input: Self::Input,
) -> Result<StreamingToolHandle<Self::OutputChunk, Self::Final>, ToolError> {
    let (chunk_tx, chunk_rx) = mpsc::channel(100);
    let (stdin_tx, mut stdin_rx) = mpsc::channel(16);
    let (done_tx, done_rx) = oneshot::channel();
    let cancel = CancellationToken::new();

    let cancel_clone = cancel.clone();
    tokio::spawn(async move {
        tokio::select! {
            _ = async {
                while let Some(event) = stdin_rx.recv().await {
                    match event {
                        StdinEvent::Text(text) => {
                            // process text input
                        }
                        StdinEvent::Signal(sig) => {
                            // handle signal
                        }
                        _ => {}
                    }
                }
            } => {}
            _ = cancel_clone.cancelled() => {}
        }
        let _ = done_tx.send(Ok(final_result));
    });

    Ok(StreamingToolHandle {
        chunks: Box::pin(ReceiverStream::new(chunk_rx)),
        stdin: Some(stdin_tx),  // bidirectional
        cancel,
        finalize: Box::pin(async move { done_rx.await.map_err(|_| ToolError::Cancelled)? }),
    })
}

Feature Flag

Streaming tools require the tools-streaming feature:

[dependencies]
aeolus = { git = "https://github.com/anemoi-ai/aeolus", tag = "v0.1.0", features = ["tools-streaming"] }

Next Steps

Getting Started — basic tool usage
Architecture — tool system internals
Security Model — WASM sandbox and capability scoping

FilesExpand file tree

streaming-tools-guide.md

Latest commit

History

streaming-tools-guide.md

File metadata and controls

Streaming Tools Guide

Overview

StreamingTool Trait

Associated Types

StreamingToolHandle

StreamingToolSession

StdinEvent

StreamingToolPolicy

1. Buffer (default)

2. Window

3. Summarise

4. Manual

Idempotency

Using Streaming Tools from an Agent

Built-in Streaming Tools

Terminal Tool (tools/terminal)

Log Tailer Tool (tools/log-tailer)

Code Interpreter Tool (tools/code-interpreter)

Implementing a Custom Streaming Tool

Read-Only Streaming Tool

Bidirectional Streaming Tool

Feature Flag

Next Steps

Terminal Tool (`tools/terminal`)

Log Tailer Tool (`tools/log-tailer`)

Code Interpreter Tool (`tools/code-interpreter`)