Bidirectional streaming for interactive Claude conversations.

Enables real-time, character-by-character streaming with persistent sessions using Claude CLI's --include-partial-messages flag.

Features

• Real-time streaming: Character-by-character text updates (typewriter effect)
• Persistent sessions: Send multiple messages without restarting subprocess
• Partial messages: Receive text_delta events as Claude types
• Multi-turn conversations: Context preserved across messages
• Phoenix LiveView ready: Perfect for chat UIs and interactive apps

Architecture

Your App
    ↓
ClaudeCodeSDK.Streaming (Public API)
    ↓
Session GenServer (Manages subprocess)
    ↓
claude --input-format stream-json \
       --output-format stream-json \
       --include-partial-messages

Quick Start

# Start a streaming session
{:ok, session} = ClaudeCodeSDK.Streaming.start_session()

# Send message and get streaming response
ClaudeCodeSDK.Streaming.send_message(session, "Hello")
|> Stream.each(fn
  %{type: :text_delta, text: text} ->
    IO.write(text)  # Print each character immediately

  %{type: :message_stop} ->
    IO.puts("")  # Message complete
end)
|> Stream.run()

# Continue conversation (context preserved)
ClaudeCodeSDK.Streaming.send_message(session, "Tell me more")
|> Enum.to_list()

# Close when done
ClaudeCodeSDK.Streaming.close_session(session)

Event Types

The stream yields event maps with these types:

Text Streaming (Primary)

• %{type: :text_delta, text: "...", accumulated: "..."} - Character chunk + full text so far
• %{type: :message_stop, final_text: "..."} - Message complete

Message Lifecycle

• %{type: :message_start, model: "...", role: "...", usage: %{}} - Message begins
• %{type: :content_block_start} - Content block starts
• %{type: :content_block_stop, final_text: "..."} - Content block complete

Tools & Thinking

• %{type: :tool_use_start, name: "...", id: "..."} - Tool call begins
• %{type: :tool_input_delta, json: "..."} - Tool input JSON chunk
• %{type: :thinking_start} - Thinking block starts (Sonnet 4.5+)
• %{type: :thinking_delta, thinking: "..."} - Thinking content chunk

Metadata & Errors

• %{type: :message_delta, stop_reason: "...", stop_sequence: "..."} - Final metadata
• %{type: :error, error: ...} - Error occurred

Examples

Typewriter Effect

{:ok, session} = ClaudeCodeSDK.Streaming.start_session()

ClaudeCodeSDK.Streaming.send_message(session, "Write a poem")
|> Stream.each(fn
  %{type: :text_delta, text: text} ->
    IO.write(text)
    Process.sleep(10)  # Slow down for effect

  %{type: :message_stop} ->
    IO.puts("\n[Complete]")
end)
|> Stream.run()

Phoenix LiveView Integration

defmodule MyAppWeb.ClaudeChatLive do
  use MyAppWeb, :live_view

  def mount(_params, _session, socket) do
    {:ok, session} = ClaudeCodeSDK.Streaming.start_session()

    {:ok, assign(socket,
      claude_session: session,
      messages: [],
      current_response: ""
    )}
  end

  def handle_event("send_message", %{"message" => message}, socket) do
    # Add user message
    messages = socket.assigns.messages ++ [%{role: :user, content: message}]

    # Start streaming in background
    pid = self()
    spawn_link(fn ->
      ClaudeCodeSDK.Streaming.send_message(socket.assigns.claude_session, message)
      |> Stream.each(fn event -> send(pid, {:claude_event, event}) end)
      |> Stream.run()
    end)

    {:noreply, assign(socket, messages: messages, current_response: "")}
  end

  def handle_info({:claude_event, %{type: :text_delta, text: text}}, socket) do
    # Update response in real-time
    {:noreply, assign(socket, current_response: socket.assigns.current_response <> text)}
  end

  def handle_info({:claude_event, %{type: :message_stop}}, socket) do
    # Finalize message
    messages = socket.assigns.messages ++ [
      %{role: :assistant, content: socket.assigns.current_response}
    ]
    {:noreply, assign(socket, messages: messages, current_response: "")}
  end
end

Multi-Turn Conversation

{:ok, session} = ClaudeCodeSDK.Streaming.start_session()

# First message
response1 = ClaudeCodeSDK.Streaming.send_message(session, "My name is Alice")
|> collect_text()

IO.puts("Claude: #{response1}")

# Second message (context preserved)
response2 = ClaudeCodeSDK.Streaming.send_message(session, "What's my name?")
|> collect_text()

IO.puts("Claude: #{response2}")  # Should mention "Alice"

ClaudeCodeSDK.Streaming.close_session(session)

defp collect_text(stream) do
  stream
  |> Stream.filter(&(&1.type == :text_delta))
  |> Stream.map(& &1.text)
  |> Enum.join("")
end

Custom Options

# Start session with custom configuration
{:ok, session} = ClaudeCodeSDK.Streaming.start_session(%ClaudeCodeSDK.Options{
  model: "opus",              # Use most capable model
  max_turns: 10,              # Allow up to 10 turns
  allowed_tools: ["Read"],    # Only allow Read tool
  verbose: true               # Enable verbose logging
})

Configuration

See ClaudeCodeSDK.Options for all available configuration options.

Performance

• Memory-efficient: Uses lazy streams, constant memory regardless of response size
• Low latency: Events delivered as soon as received from Claude
• Concurrent: Multiple sessions can run simultaneously
• Resource cleanup: Subprocess terminated on session close

Error Handling

ClaudeCodeSDK.Streaming.send_message(session, "Hello")
|> Stream.each(fn
  %{type: :text_delta, text: text} ->
    IO.write(text)

  %{type: :error, error: reason} ->
    IO.puts("\nError: #{inspect(reason)}")

  %{type: :message_stop} ->
    IO.puts("\nComplete")
end)
|> Stream.run()

Limitations

• Requires Claude CLI v2.0+ with streaming support
• Subprocess spawning has ~100ms overhead
• No message history persistence (use ClaudeCodeSDK.SessionStore for that)
• Single active message per session (sequential only)

See Also

• ClaudeCodeSDK.Streaming.Session - GenServer managing the subprocess
• ClaudeCodeSDK.Streaming.EventParser - Event parsing logic
• ClaudeCodeSDK.Options - Configuration options
• ClaudeCodeSDK - Non-streaming API (simpler but blocks)