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

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 === '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');
session.close();

import os

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()


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).
• Both SDKs expose the canonical event.type; Python retains event.event_type as a compatibility alias. event.payload is the complete lossless payload, and unknown future types are preserved. Convenience 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.