Event Streaming

Event streaming lets applications observe model output, tool calls, graph state, custom updates, subgraph activity, and completion metadata while an agent or graph runs.

BeamWeaver's event stream is an Enumerable of typed %BeamWeaver.Stream.Envelope{} values. Each envelope carries:

• event: a typed event struct from BeamWeaver.Stream.Events

• run_id, graph, node, task_id, and step

• namespace: the current graph or subgraph path

• metadata: provider, tracing, usage, tags, or app-specific metadata

• timestamp

BeamWeaver.Stream.Events is the documented namespace for stream event structs such as Token, Message, ToolStart, GraphValue, and Done. Pattern match on those structs when you need raw stream detail, or pass the same event list through projection reducers when you need messages, values, subgraph summaries, or lifecycle views.

For agents, use BeamWeaver.Agent.stream_events/3. For compiled graphs, use BeamWeaver.Graph.Compiled.stream_events/3. For standalone chat models, use BeamWeaver.Core.ChatModel.stream_typed_events/3 when you want normalized BeamWeaver envelopes directly. Provider modules can also expose provider-specific stream_events/3 lifecycle streams; use those when you need the raw provider semantic lifecycle rather than the normalized BeamWeaver event contract.

Standalone OpenAI, Anthropic, Google, DeepSeek, xAI, and Z.ai model streams are lazy live enumerables when using the live transport. Provider chunks are parsed incrementally as server-sent events arrive. In tests, fake or replay transports can emit deterministic typed stream events from fixtures. If a lazy provider stream fails before any model output is emitted, consumers see an %BeamWeaver.Stream.Events.Error{} item when they enumerate the stream. Raw provider lazy-stream helpers can also accept an on_response callback when the caller needs transport status and headers without changing the lazy return type.

Quickstart

Build or define an agent, then stream typed events:

alias BeamWeaver.Agent
alias BeamWeaver.Core.{Message, Tool}
alias BeamWeaver.Stream.Envelope
alias BeamWeaver.Stream.Events

weather_tool =
  Tool.from_function!(
    name: "get_weather",
    description: "Get weather for a city.",
    input_schema: %{
      "type" => "object",
      "properties" => %{"city" => %{"type" => "string"}},
      "required" => ["city"]
    },
    handler: fn %{"city" => city}, _opts ->
      "It's always sunny in #{city}!"
    end
  )

{:ok, agent} =
  Agent.build(
    model: BeamWeaver.Models.init_chat_model!("openai:gpt-5-nano"),
    tools: [weather_tool],
    name: "weather_agent"
  )

{:ok, events} =
  Agent.stream_events(
    agent,
    %{messages: [Message.user("What is the weather in SF?")]},
    live: true
  )

for %Envelope{event: event} <- events do
  case event do
    %Events.Token{text: delta} ->
      IO.write(delta)

    %Events.Message{message: message} ->
      IO.write(Message.text(message))

    %Events.ToolStart{tool_name: name, input: input} ->
      IO.inspect({:tool_start, name, input})

    %Events.ToolFinish{tool_call_id: id, output: output} ->
      IO.inspect({:tool_finish, id, output})

    %Events.ToolError{tool_call_id: id, message: message} ->
      IO.inspect({:tool_error, id, message})

    _other ->
      :ok
  end
end

Use live: true when a UI needs events as the run progresses. Live streams use BeamWeaver.Stream.Mux, so completion is visible as producer lifecycle/debug events. Without live: true, graph and agent streams are collected by the runtime before they are returned, which is useful for tests and projection passes and includes a terminal %Events.Done{} when using Compiled.stream_events/3.

How The Pieces Fit Together

BeamWeaver has the same two conceptual layers as LangGraph's event-streaming docs, but the public surface is Elixir-shaped:

1. Graph execution emits typed envelopes while nodes, tools, checkpoints, and subgraphs run.

2. Projection reducers such as BeamWeaver.Stream.MessagesTransformer, BeamWeaver.Stream.Transformers, BeamWeaver.Stream.Subgraphs, and BeamWeaver.Stream.Lifecycle derive application views from those envelopes.

flowchart TD
  A["Graph or agent runtime"] --> B["Typed envelope stream"]
  B --> C["Message reducer"]
  B --> D["State/mode reducer"]
  B --> E["Subgraph reducer"]
  B --> F["Application reducer"]
  C --> G["Message streams"]
  D --> H["Values, updates, tasks, checkpoints"]
  E --> I["Subgraph summaries"]
  F --> J["UI or telemetry projection"]

The important difference is ownership. LangGraph's Python run object owns live projection handles such as stream.messages, stream.values, and stream.output. BeamWeaver returns one typed Enumerable. Application code can consume it live in one pass, or collect a bounded event list and run as many immutable projections as needed.

What You Can Stream

Projection Map

LangGraph's official page describes projections on the Python run object. BeamWeaver maps those to typed events and reducers:

Protocol Envelope And Channels

BeamWeaver does not expose LangGraph's ProtocolEvent dictionaries as the primary API. The native protocol envelope is %BeamWeaver.Stream.Envelope{}:

%BeamWeaver.Stream.Envelope{
  event: %BeamWeaver.Stream.Events.GraphValue{value: state},
  run_id: "run-1",
  graph: "ResearchGraph",
  node: :draft,
  task_id: "task-1",
  step: 2,
  namespace: ["researcher"],
  metadata: %{},
  timestamp: 1_768_000_000_000_000
}

The event struct determines the logical channel:

Use BeamWeaver.Stream.event_mode/1 when you need to classify an event programmatically:

for envelope <- events do
  IO.inspect({BeamWeaver.Stream.event_mode(envelope.event), envelope.namespace})
end

Envelope namespaces are already parsed into a list. The root graph uses []. Nested graph execution appends child node names, so a tool event inside a child graph can arrive with namespace: ["researcher", "tools"].

Agent Messages

For live UI output, consume token and message events directly:

alias BeamWeaver.Core.Message
alias BeamWeaver.Stream.Events

for envelope <- events do
  case envelope.event do
    %Events.Token{text: delta} ->
      IO.write(delta)

    %Events.Message{message: message} ->
      IO.write(Message.text(message))

    %Events.MessageChunk{chunk: chunk} ->
      IO.inspect(chunk, label: "message chunk")

    _other ->
      :ok
  end
end

When you want a message-level projection after a run, use BeamWeaver.Stream.MessagesTransformer:

alias BeamWeaver.Stream.{MessageStream, MessagesTransformer}

event_list = Enum.to_list(events)

{:ok, transformer, _emitted} =
  MessagesTransformer.process_many(MessagesTransformer.new(), event_list)

for message_stream <- MessagesTransformer.streams(transformer) do
  IO.puts("[#{message_stream.node}] " <> MessageStream.text(message_stream))

  {:ok, message} = MessageStream.output(message_stream)
  IO.inspect(message.usage_metadata, label: "usage")
end

Use :pre_projection when a transform must mutate typed stream envelopes before they become message projections. This is intentionally narrow and pure: each function receives one event and returns the event, {:ok, event}, :drop, or {:error, reason}.

alias BeamWeaver.Agent.Middleware.PII
alias BeamWeaver.Stream.{MessageStream, MessagesTransformer}

transformer =
  MessagesTransformer.new(
    pre_projection: PII.stream_transform(type: :email, strategy: :redact)
  )

{:ok, transformer, _emitted} =
  MessagesTransformer.process_many(transformer, event_list)

for stream <- MessagesTransformer.streams(transformer) do
  IO.puts(MessageStream.text(stream))
end

Provider Stream Bounds

Live provider SSE streams validate decoded events before publishing them. The defaults allow 100,000 events, 64 MiB of encoded transport data, and 64 MiB of decoded event values. :max_stream_events, :max_stream_bytes, and :max_stream_value_bytes lower those limits; :max_bytes, :max_items, and :max_depth bound each decoded value.

Validation is batch-atomic and sticky. If decoding or validation fails, no event from that candidate batch is emitted, the first error is retained, and the stream cannot later finish successfully. Collected non-live responses use :max_response_bytes, which defaults to 16 MiB.

Provider adapters use BeamWeaver.Provider.StreamValidator internally. Custom adapters can use the same policy-neutral state directly:

alias BeamWeaver.Provider.StreamValidator

validation = StreamValidator.new(max_stream_events: 10_000)

with {:ok, validation} <-
       StreamValidator.push(validation, decoded_events,
         transport_bytes: byte_size(chunk)
       ),
     :ok <- StreamValidator.finish(validation) do
  decoded_events
end

Reasoning Content

Reasoning output is provider-dependent. OpenAI, Anthropic, Google, DeepSeek, xAI, and Z.ai can surface reasoning as content blocks in %Events.MessageChunk{} events. Treat those chunk events as the live thinking stream; final assistant messages and text projections should be considered answer/tool output, not the primary place to read incremental reasoning.

alias BeamWeaver.Stream.Events

for envelope <- events do
  case envelope.event do
    %Events.MessageChunk{chunk: %{content: blocks}} when is_list(blocks) ->
      for block <- blocks do
        case block do
          %{"type" => "reasoning", "text" => text} ->
            IO.write("[thinking] " <> text)

          %{"type" => "reasoning", "reasoning" => text} ->
            IO.write("[thinking] " <> text)

          %{type: :reasoning, reasoning: text} ->
            IO.write("[thinking] " <> text)

          _other ->
            :ok
        end
      end

    _other ->
      :ok
  end
end

Tool Calls

There are two tool-related event families:

• %Events.ToolCallChunk{} for model-generated tool-call argument chunks

• %Events.ToolStart{}, %Events.ToolDelta{}, %Events.ToolFinish{}, and %Events.ToolError{} for tool execution lifecycle

Model tool-call chunks:

alias BeamWeaver.Stream.Events

for envelope <- events do
  case envelope.event do
    %Events.ToolCallChunk{chunk: chunk} ->
      IO.inspect(
        %{id: chunk.id, name: chunk.name, args: chunk.args, index: chunk.index},
        label: "tool call chunk"
      )

    _other ->
      :ok
  end
end

OpenAI and xAI streams preserve role-only empty chunks and incremental tool-call argument chunks. Consumers should treat empty chunks as lifecycle evidence rather than text, and should rely on the final %Events.Message{} or stream_response/3 result for the reconstructed assistant tool calls. See examples/streaming_tool_call_chunks.exs for an offline reconstruction example.

To reconstruct finalized tool calls from streamed message chunks, collect the chunks and finalize them:

chunks =
  for %{event: %Events.MessageChunk{chunk: chunk}} <- event_list do
    chunk
  end

final_message = BeamWeaver.Stream.Finalize.finalize(chunks)
IO.inspect(final_message.tool_calls, label: "final tool calls")

Tool execution lifecycle:

for envelope <- events do
  case envelope.event do
    %Events.ToolStart{tool_call_id: id, tool_name: name, input: input} ->
      IO.inspect({:started, id, name, input})

    %Events.ToolDelta{tool_call_id: id, delta: delta} ->
      IO.inspect({:delta, id, delta})

    %Events.ToolFinish{tool_call_id: id, output: output} ->
      IO.inspect({:finished, id, output})

    %Events.ToolError{tool_call_id: id, message: message, error_type: type} ->
      IO.inspect({:failed, id, type, message})

    _other ->
      :ok
  end
end

Tools can emit output deltas through injected tool runtime:

alias BeamWeaver.Core.{Tool, ToolRuntime}

streaming_tool =
  Tool.from_function!(
    name: "streaming_echo",
    description: "Echo chunks while running.",
    input_schema: %{
      "type" => "object",
      "properties" => %{"text" => %{"type" => "string"}},
      "required" => ["text"]
    },
    injected: %{"tool_runtime" => :tool_runtime},
    handler: fn %{"text" => text, "tool_runtime" => runtime}, _opts ->
      ToolRuntime.emit_output_delta(runtime, "starting")
      ToolRuntime.emit_output_delta(runtime, "finishing")
      text
    end
  )

Deep Agents Task Subagents

Deep Agents task subagents are delegated through the task tool. LangChain's Python SDK exposes these as stream.subagents handles with nested message, tool-call, value, and output projections. BeamWeaver exposes the same product concept as an immutable read-side projection over typed events: BeamWeaver.Agent.Subagent.StreamTransformer.

Use the transformer when you want user-facing subagent cards, telemetry, or summaries for delegated work:

alias BeamWeaver.Agent
alias BeamWeaver.Agent.Subagent.{RunStream, StreamTransformer}

{:ok, events} = Agent.stream_events(agent, input)

{:ok, transformer, _new_handles} =
  StreamTransformer.process_many(
    StreamTransformer.new(subagent_names: ["researcher", "coder"]),
    events
  )

transformer = StreamTransformer.finalize(transformer)

for handle <- transformer.log do
  IO.inspect(
    %{
      name: RunStream.name(handle),
      path: handle.path,
      status: handle.status,
      task_input: handle.task_input,
      cause: RunStream.cause(handle),
      output: RunStream.output(handle)
    },
    label: "subagent"
  )
end

Declare the same names that appear in the agent's subagents configuration. Undeclared names are ignored so ordinary tool calls and internal graph tasks do not accidentally appear as Deep Agents subagents.

Subagent Stream Fields

BeamWeaver.Agent.Subagent.RunStream stores the data most applications need from a delegated run:

BeamWeaver also keeps task_input, the task description passed into the child, and RunStream.cause/1, a map with the parent tool-call ID when the subagent was started by a visible task call.

Track Lifecycle

You can observe newly discovered subagent handles while consuming the stream in arrival order:

alias BeamWeaver.Agent.Subagent.{RunStream, StreamTransformer}

transformer =
  Enum.reduce(events, StreamTransformer.new(subagent_names: ["researcher"]), fn envelope, acc ->
    case StreamTransformer.process(acc, envelope) do
      {:ok, next, new_handles} ->
        Enum.each(new_handles, fn handle ->
          IO.puts("#{RunStream.name(handle)} started at #{Enum.join(handle.path, "/")}")
        end)

        next

      {:pass, next} ->
        next
    end
  end)
  |> StreamTransformer.finalize()

for handle <- transformer.log do
  IO.puts("#{RunStream.name(handle)} #{handle.status}")
end

For live UI updates, consume the event enumerable once and route each envelope to your UI state, a GenServer, or PubSub topic. BeamWeaver does not expose independent live iterators such as stream.messages and stream.subagents; multiple consumers should be explicit application processes or read-side projections over a collected event list.

Stream Messages And Tool Calls

Subagent message and tool-call streams are just the events observed under the child namespace:

alias BeamWeaver.Stream.Events

for handle <- transformer.log do
  for %{event: %Events.Message{message: message}} <- handle.events do
    IO.inspect(message, label: "#{handle.graph_name} message")
  end

  for %{event: %Events.ToolStart{tool_name: name, input: input}} <- handle.events do
    IO.inspect({name, input}, label: "#{handle.graph_name} tool")
  end

  for %{event: %Events.ToolDelta{delta: delta}} <- handle.events do
    IO.inspect(delta, label: "#{handle.graph_name} tool delta")
  end
end

If you need exact arrival order across the coordinator and every subagent, keep the raw envelope stream as your source of truth and branch on envelope.namespace. The transformer is intended for Deep Agents task summaries and scoped event views.

Nested Subagents

Nested task delegations appear in handle.subagents:

alias BeamWeaver.Agent.Subagent.RunStream

print_subagents = fn handles ->
  Enum.each(handles, fn handle ->
    IO.puts("#{RunStream.name(handle)} #{handle.status}")

    Enum.each(handle.subagents, fn nested ->
      IO.puts("  #{RunStream.name(nested)} #{nested.status}")
    end)
  end)
end

print_subagents.(transformer.log)

For deeper trees, recurse through handle.subagents using ordinary Elixir data traversal.

Subgraphs Versus Subagents

Use StreamTransformer for product-level Deep Agents task delegations. Use BeamWeaver.Stream.Subgraphs for static graph structure, named graph nodes, and compiled subgraphs.

Subgraph events carry a namespace. While consuming events live, filter by envelope.namespace:

for envelope <- events do
  case envelope.namespace do
    ["weather_agent" | _rest] ->
      IO.inspect(envelope.event, label: "weather agent")

    _other ->
      :ok
  end
end

After collecting events, project subgraph summaries:

alias BeamWeaver.Stream.Subgraphs

runs = Subgraphs.from_events(event_list)

for subgraph <- Subgraphs.flatten(runs) do
  IO.inspect(
    %{
      graph_name: subgraph.graph_name,
      path: subgraph.path,
      status: subgraph.status,
      values: subgraph.values
    },
    label: "subgraph"
  )
end

You can also project lifecycle envelopes:

alias BeamWeaver.Stream.Lifecycle

for lifecycle_envelope <- Lifecycle.from_events(event_list) do
  IO.inspect(lifecycle_envelope.event)
end

State And Final Output

Use graph value events when you need state snapshots:

alias BeamWeaver.Stream.Events

final_state =
  event_list
  |> Enum.reduce(nil, fn
    %{event: %Events.GraphValue{value: value}}, _acc -> value
    _event, acc -> acc
  end)

You can also project values from a collected event list:

alias BeamWeaver.Stream.Transformers

values =
  event_list
  |> Transformers.stream(:values)
  |> Enum.map(fn {_mode, envelope} -> envelope end)

final_state =
  values
  |> Enum.reduce(nil, fn
    %{event: %Events.GraphValue{value: value}}, _acc -> value
    _event, acc -> acc
  end)

If you only need the final state, prefer BeamWeaver.Agent.invoke/3. Event streams are for progress, observability, and UI updates.

Resume After An Interrupt

When a graph pauses for human input, non-live event streaming returns an interrupt tuple rather than a Python run object with stream.interrupted:

alias BeamWeaver.Graph
alias BeamWeaver.Graph.Compiled

graph = Graph.compile!(workflow, checkpointer: checkpointer)
config = %{"configurable" => %{"thread_id" => "review-1"}}

case Compiled.stream_events(graph, input, config: config) do
  {:ok, events} ->
    Enum.each(events, &IO.inspect(&1.event))

  {:interrupted, %{events: events} = interrupt} ->
    Enum.each(events, &IO.inspect(&1.event))
    IO.inspect(interrupt.value, label: "review payload")
end

Resume with BeamWeaver.Graph.Compiled.resume/3 or BeamWeaver.Agent.resume/3 using the same checkpointer and thread_id:

{:ok, final_state} =
  Compiled.resume(graph, %{decisions: [%{type: :approve}]}, config: config)

For live streams, interrupt information is emitted as a debug envelope:

alias BeamWeaver.Stream.Events

for %{event: %Events.Debug{payload: %{type: :interrupt, interrupt: interrupt}}} <- events do
  IO.inspect(interrupt, label: "paused")
end

Resume still uses resume/3; there is no BeamWeaver equivalent of calling stream_events(Command(...), version="v3").

Multiple Projections

For one-pass live consumption, dispatch on the typed event:

Enum.reduce(events, %{text: "", tool_calls: %{}, values: []}, fn envelope, acc ->
  case envelope.event do
    %Events.Token{text: delta} ->
      %{acc | text: acc.text <> delta}

    %Events.ToolStart{tool_call_id: id, tool_name: name, input: input} ->
      put_in(acc, [:tool_calls, id], %{name: name, input: input, deltas: []})

    %Events.ToolDelta{tool_call_id: id, delta: delta} ->
      tool_calls =
        Map.update(acc.tool_calls, id, %{deltas: [delta]}, fn call ->
          Map.update(call, :deltas, [delta], &(&1 ++ [delta]))
        end)

      %{acc | tool_calls: tool_calls}

    %Events.GraphValue{value: value} ->
      %{acc | values: acc.values ++ [value]}

    _other ->
      acc
  end
end)

For collected events, use projection reducers:

alias BeamWeaver.Stream.Transformers

for {mode, envelope} <- Transformers.stream(event_list, [:values, :tasks, :lifecycle]) do
  IO.inspect({mode, envelope.event})
end

Build Your Own Projection

LangGraph's page introduces StreamTransformer, StreamChannel, required_stream_modes, named channels, unnamed channels, and compile-time or call-time transformer registration. BeamWeaver does not expose those Python classes. The native pattern is to write a small reducer over typed envelopes:

defmodule MyApp.ToolActivityProjection do
  alias BeamWeaver.Stream.Events

  def reduce(events) do
    Enum.reduce(events, %{}, &step/2)
  end

  def step(%{event: %Events.ToolStart{tool_call_id: id, tool_name: name, input: input}}, acc) do
    Map.put(acc, id, %{name: name, input: input, status: :started, deltas: []})
  end

  def step(%{event: %Events.ToolDelta{tool_call_id: id, delta: delta}}, acc) do
    Map.update(acc, id, %{status: :streaming, deltas: [delta]}, fn call ->
      Map.update(call, :deltas, [delta], &(&1 ++ [delta]))
    end)
  end

  def step(%{event: %Events.ToolFinish{tool_call_id: id, output: output}}, acc) do
    Map.update(acc, id, %{status: :finished, output: output}, fn call ->
      Map.merge(call, %{status: :finished, output: output})
    end)
  end

  def step(%{event: %Events.ToolError{tool_call_id: id, message: message, error_type: type}}, acc) do
    Map.update(acc, id, %{status: :error, error: {type, message}}, fn call ->
      Map.merge(call, %{status: :error, error: {type, message}})
    end)
  end

  def step(_event, acc), do: acc
end

For live output, use Stream.transform/3 to emit derived values as the base event stream is consumed:

activity_stream =
  Stream.transform(events, %{}, fn envelope, acc ->
    next = MyApp.ToolActivityProjection.step(envelope, acc)
    {[next], next}
  end)

For fan-in from multiple live producers, use BeamWeaver.Stream.Mux. For application-specific progress, emit %Events.Custom{} values and reduce them like any other event.

Custom Updates

Graph nodes can emit application-specific updates through runtime.stream_writer:

alias BeamWeaver.Graph
alias BeamWeaver.Graph.Compiled
alias BeamWeaver.Stream.Events

graph =
  Graph.new()
  |> Graph.add_node(:retrieve, fn _state, runtime ->
    runtime.stream_writer.(%{phase: :retrieving, progress: 0.25})
    runtime.stream_writer.(%{phase: :reranking, progress: 0.75})
    %{answer: "done"}
  end)
  |> Graph.add_edge(Graph.start(), :retrieve)
  |> Graph.add_edge(:retrieve, Graph.end_node())
  |> Graph.compile!()

{:ok, events} = Compiled.stream_events(graph, %{})

for %{event: %Events.Custom{payload: payload}} <- events do
  IO.inspect(payload, label: "custom update")
end

Because events are ordinary Elixir values, custom projections can be ordinary functions, reducers, or Stream.map/2 pipelines. See examples/streaming_redaction.exs for a credential-free pre-projection redaction example.