Streaming

session.stream(prompt) yields incremental events as the turn runs, so you can render text as it arrives and react to tool activity in real time. Use it when you want a live UI or a CLI that prints output token-by-token instead of waiting for the full result from send or run.

Each event carries the stable envelope fields version, type, payload, and optional metadata. The common kinds are agent_start, text_delta / reasoning_delta, tool_start / tool_end, agent_end, and error. Verification data is carried by the agent_end payload and convenience fields.

Rust

Node.js

Python

Rust

use a3s_code_core::{Agent, AgentEvent, CodeError, PlanningMode, SessionOptions};

#[tokio::main]
async fn main() -> a3s_code_core::Result<()> {
    let agent = Agent::new("agent.acl").await?;
    let session = agent
        .session_builder(".")
        .options(SessionOptions::new().with_planning_mode(PlanningMode::Disabled))
        .build()
        .await?;

    let (mut events, lifecycle) = session
        .stream(
            "Use the bash tool to run the tests, then summarize the result.",
            None,
        )
        .await?;

    while let Some(event) = events.recv().await {
        match event {
            AgentEvent::TextDelta { text } => print!("{text}"),
            AgentEvent::ToolStart { name, .. } => println!("\n[tool:start] {name}"),
            AgentEvent::ToolEnd {
                name, exit_code, ..
            } => println!("\n[tool:end] {name} exit={exit_code}"),
            AgentEvent::End {
                verification_summary,
                ..
            } => println!("\n[verification] {verification_summary:?}"),
            AgentEvent::Error { message } => return Err(CodeError::Llm(message)),
            _ => {}
        }
    }
    lifecycle
        .await
        .map_err(|error| CodeError::Internal(error.into()))?;
    println!("\n[stream] complete");

    session.close().await;
    agent.close().await;
    Ok(())
}

TypeScript

import { Agent } from '@a3s-lab/code';

const agent = await Agent.create('agent.acl');
const session = await agent.sessionAsync(process.cwd(), {
  planningMode: 'disabled',
});

const stream = await session.stream(
  'Use the bash tool to run the tests, then summarize the result.',
);

for await (const event of stream) {
  if (event.type === 'text_delta' && event.text) {
    process.stdout.write(event.text);
  } else if (event.type === 'tool_start') {
    console.log(`\n[tool:start] ${event.toolName ?? 'unknown'}`);
  } else if (event.type === 'tool_end') {
    console.log(
      `\n[tool:end] ${event.toolName ?? 'unknown'} exit=${event.exitCode ?? 0}`,
    );
  } else if (event.type === 'agent_end') {
    console.log(`\n[verification] ${event.verificationSummaryText ?? ''}`);
  } else if (event.type === 'error') {
    throw new Error(event.error ?? 'stream error');
  }
}

console.log('\n[stream] complete');
await session.closeAsync();
await agent.close();

Python

from a3s_code import Agent, SessionOptions


def main() -> None:
    agent = Agent.create("agent.acl")

    opts = SessionOptions()
    opts.planning_mode = "disabled"
    session = agent.session(".", opts)

    prompt = "Use the bash tool to run the tests, then summarize the result."

    try:
        for event in session.stream(prompt):
            if event.type == "text_delta" and event.text:
                print(event.text, end="", flush=True)
            elif event.type == "tool_start":
                print(f"\n[tool:start] {event.tool_name or 'unknown'}")
            elif event.type == "tool_end":
                print(f"\n[tool:end] {event.tool_name or 'unknown'} exit={event.exit_code or 0}")
            elif event.type == "agent_end":
                print(f"\n[verification] {event.verification_summary_text or ''}")
            elif event.type == "error":
                raise RuntimeError(event.error or "stream error")
        print("\n[stream] complete")
    finally:
        session.close()
        agent.close()


if __name__ == "__main__":
    main()

package main

import (
	"context"
	"fmt"
	"log"

	code "github.com/A3S-Lab/Code/sdk/go/v9"
)

func main() {
	ctx := context.Background()
	agent, err := code.Create(ctx, "agent.acl")
	if err != nil {
		log.Fatal(err)
	}
	defer agent.Close(context.Background())

	session, err := agent.Session(ctx, ".", &code.SessionOptions{
		PlanningMode: code.PlanningDisabled,
	})
	if err != nil {
		log.Fatal(err)
	}
	defer session.Close(context.Background())

	stream, err := session.Stream(
		ctx,
		"Use the bash tool to run the tests, then summarize the result.",
		nil,
	)
	if err != nil {
		log.Fatal(err)
	}
	for event := range stream.Events {
		// Payload fields follow the wire event: text, name, exit_code,
		// verification_summary, message.
		var payload struct {
			Text                string                   `json:"text"`
			Name                string                   `json:"name"`
			ExitCode            int                      `json:"exit_code"`
			VerificationSummary code.VerificationSummary `json:"verification_summary"`
			Message             string                   `json:"message"`
		}
		switch event.Type {
		case code.EventTextDelta, code.EventToolStart, code.EventToolEnd,
			code.EventAgentEnd, code.EventError:
			if err := event.DecodePayload(&payload); err != nil {
				log.Fatal(err)
			}
		default:
			// Forward other events (event.Payload, event.Metadata) as needed.
			continue
		}
		switch event.Type {
		case code.EventTextDelta:
			fmt.Print(payload.Text)
		case code.EventToolStart:
			fmt.Printf("\n[tool:start] %s\n", payload.Name)
		case code.EventToolEnd:
			fmt.Printf("\n[tool:end] %s exit=%d\n", payload.Name, payload.ExitCode)
		case code.EventAgentEnd:
			fmt.Printf("\n[verification] %s\n", payload.VerificationSummary.Status)
		case code.EventError:
			log.Fatal(payload.Message)
		}
	}
	if err := <-stream.Done; err != nil {
		log.Fatal(err)
	}
	fmt.Println("\n[stream] complete")
}

Notes:

Rust receives AgentEvent values from a Tokio channel and awaits the returned lifecycle handle. Node.js iterates the EventStream with for await (or calls stream.next() directly). Python's EventStream supports both for and async for. Go drains stream.Events before reading the terminal error from stream.Done. Canceling the Go context also requests cancellation of the active native run.
All four SDKs expose the canonical event type. Node.js and Python add convenience projections; Go keeps Payload and Metadata as json.RawMessage and provides DecodePayload. Unknown future event types remain lossless in every SDK.
tool_start fires when the model begins emitting a tool call; tool_execution_start fires when the tool actually runs, and tool_end carries its output and exit code.
With a confirmation policy enabled, the stream also carries human-in-the-loop signals: confirmation_required, then confirmation_received once an answer arrives. A pending confirmation stays parked until it is answered; no timer settles it. confirmation_timeout is emitted, with action_taken, only when the confirmation channel closes before an answer (for example, on cancel or close).

Runnable streaming examples ship under sdk/node/examples/streaming/. A complete human-in-the-loop confirmation loop ships at sdk/node/examples/streaming/hitl_confirmation_loop.ts, with the matching Python version under sdk/python/examples/ and Go stream coverage in sdk/go/session_test.go.

#Streaming

Streaming