Streaming

The Amp SDK uses a streaming architecture to deliver messages in real time as the agent works.

How It Works

AmpSdk.execute/2 returns a lazy Stream that yields typed message structs. The stream is backed by Stream.resource/3 — messages are produced from the shared core session API and projected back into the public Amp SDK message types, and the stream halts automatically when a final result or error is received.

The stream consumer uses selective receive for the internal runtime messages that belong to the active stream reference only. This means unrelated mailbox messages in the caller process are preserved, so execute/2 can be used safely from OTP processes that also handle other messages.

When a stream finishes (result, timeout, parse error, or transport error), cleanup drains any remaining internal runtime messages for that session reference. This avoids leaving stale stream events in long-lived caller mailboxes.

execute/2 accepts either:

• a string prompt (--stream-json / --stream-json-thinking)
• a list of UserInputMessage values (--stream-json-input)

The public stream message structs are schema-backed. Known fields are normalized through Zoi, forward-compatible unknown fields are preserved in extra, and the message/content modules expose to_map/1 for projection back to wire shape.

Your App  →  AmpSdk.execute/2  →  AmpSdk.Stream  →  AmpSdk.Runtime.CLI  →  cli_subprocess_core  →  amp CLI
                                     (Stream.resource)      (session kit)        (shared runtime)      (subprocess)

Message Types

Messages are yielded in this order during a typical execution:

1. SystemMessage

Emitted once when the provider session is established.

%AmpSdk.Types.SystemMessage{
  type: "system",
  subtype: "init",
  session_id: "T-abc123...",
  cwd: "/path/to/project",
  tools: ["Bash", "Read", "edit_file", ...],
  mcp_servers: [%{name: "fs", status: "connected"}]
}

2. AssistantMessage

The agent's response — may contain text and/or tool calls.

%AmpSdk.Types.AssistantMessage{
  type: "assistant",
  session_id: "T-abc123...",
  message: %{
    role: "assistant",
    content: [
      %AmpSdk.Types.TextContent{type: "text", text: "Here's what I found..."},
      %AmpSdk.Types.ToolUseContent{type: "tool_use", id: "tu_1", name: "Bash", input: %{"command" => "ls"}}
    ],
    stop_reason: "end_turn",  # or "tool_use"
    usage: %AmpSdk.Types.Usage{input_tokens: 100, output_tokens: 50}
  }
}

Thinking Content (with Options.thinking: true)

When thinking: true is set, assistant messages may contain ThinkingContent blocks:

%AmpSdk.Types.ThinkingContent{type: "thinking", thinking: "Let me work through this..."}

These appear before text content in the content list and represent the model's chain-of-thought reasoning.

3. UserMessage

Tool results fed back to the agent (generated by the CLI, not by you).

%AmpSdk.Types.UserMessage{
  type: "user",
  message: %{
    role: "user",
    content: [
      %AmpSdk.Types.ToolResultContent{tool_use_id: "tu_1", content: "file.txt\nREADME.md"}
    ]
  }
}

4. ResultMessage / ErrorResultMessage

Final outcome — the stream halts after this.

%AmpSdk.Types.ResultMessage{
  result: "The project has 12 modules.",
  duration_ms: 3500,
  num_turns: 2,
  is_error: false
}

%AmpSdk.Types.ErrorResultMessage{
  error: "Permission denied for tool Bash",
  duration_ms: 100,
  num_turns: 0,
  is_error: true,
  permission_denials: ["Bash"]
}

Patterns

Collect Final Result Only

{:ok, result} = AmpSdk.run("What is 2+2?")

Typewriter Effect

AmpSdk.execute("Write a poem")
|> Enum.each(fn
  %AmpSdk.Types.AssistantMessage{message: %{content: content}} ->
    for %AmpSdk.Types.TextContent{text: text} <- content, do: IO.write(text)

  %AmpSdk.Types.ResultMessage{} ->
    IO.puts("\n--- done ---")

  _ -> :ok
end)

Accumulate with Metadata

result =
  AmpSdk.execute("Analyze this code")
  |> Enum.reduce(%{text: "", turns: 0, ms: 0}, fn
    %AmpSdk.Types.AssistantMessage{message: %{content: content}}, acc ->
      text = content |> Enum.filter(&match?(%{type: "text"}, &1)) |> Enum.map(& &1.text) |> Enum.join()
      %{acc | text: acc.text <> text}

    %AmpSdk.Types.ResultMessage{num_turns: t, duration_ms: ms}, acc ->
      %{acc | turns: t, ms: ms}

    _, acc -> acc
  end)

Filter by Message Type

tool_calls =
  AmpSdk.execute("Fix this bug")
  |> Stream.filter(&match?(%AmpSdk.Types.AssistantMessage{}, &1))
  |> Stream.flat_map(fn %{message: %{content: content}} ->
    Enum.filter(content, &match?(%AmpSdk.Types.ToolUseContent{}, &1))
  end)
  |> Enum.to_list()

Timeouts

Set stream timeout directly via Options.stream_timeout_ms:

alias AmpSdk.Types.Options

AmpSdk.execute("Long running task", %Options{stream_timeout_ms: 30_000})
|> Enum.to_list()

For outer cancellation boundaries, you can still wrap with Task:

task = Task.async(fn ->
  AmpSdk.execute("Long running task") |> Enum.to_list()
end)

case Task.yield(task, 30_000) || Task.shutdown(task) do
  {:ok, messages} -> handle_messages(messages)
  nil -> IO.puts("Timed out")
end