amp_sdk

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

```elixir
%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.

```elixir
%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:

```elixir
%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).

```elixir
%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.

```elixir
%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

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

### Typewriter Effect

```elixir
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

```elixir
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

```elixir
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`:

```elixir
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`:

```elixir
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
```