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 a type discriminant. The common kinds are text_delta (a chunk of assistant text), tool_start / tool_end (a tool call beginning and finishing), verification (a verification summary), and error.

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

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

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

while (true) {
  const next = await stream.next();
  if (next.done || !next.value) break;

  const event = next.value;
  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 === 'verification') {
    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.close();

import os

from a3s_code import Agent, SessionOptions


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

    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.event_type == "text_delta" and event.text:
                print(event.text, end="", flush=True)
            elif event.event_type == "tool_start":
                print(f"\n[tool:start] {event.tool_name or 'unknown'}")
            elif event.event_type == "tool_end":
                print(f"\n[tool:end] {event.tool_name or 'unknown'} exit={event.exit_code or 0}")
            elif event.event_type == "verification":
                print(f"\n[verification] {event.verification_summary_text or ''}")
            elif event.event_type == "error":
                raise RuntimeError(event.error or "stream error")
        print("\n[stream] complete")
    finally:
        session.close()


if __name__ == "__main__":
    main()

Notes:

Node iterates the stream manually with stream.next(), checking next.done and next.value. In the current build the Python SDK exposes streaming as a synchronous iterator, so you consume it with a plain for loop (orchestration APIs such as parallel and pipeline remain async).
The type discriminant differs by language: Node reads event.type, Python reads event.event_type. Other fields follow each language's casing: toolName / exitCode / verificationSummaryText in Node, tool_name / exit_code / verification_summary_text in Python.
Streamed events can also include human-in-the-loop confirmation signals (confirmation_required, confirmation_received, confirmation_timeout) when a confirmation policy is enabled.

Runnable streaming examples ship under crates/code/sdk/node/examples/streaming/. A complete human-in-the-loop confirmation loop ships at crates/code/sdk/node/examples/streaming/hitl_confirmation_loop.ts, and the matching Python version ships at crates/code/sdk/python/examples/hitl_confirmation_loop.py.

Streaming

Streaming

On this page