Claude API in Elixir: Streaming, Tool Use, and Structured Output in Phoenix

Every “how do I call an LLM” tutorial assumes you are in Python or TypeScript, because that is where the official SDKs live. Anthropic ships SDKs for Python, TypeScript, Java, Go, Ruby, C# and PHP. It does not ship one for Elixir. So when you go looking for how to use the Claude API in Elixir, you find either a thin community wrapper you have to audit yourself, or nothing at all.

That gap is smaller than it looks. The Claude Messages API is a single HTTP endpoint that takes JSON and returns JSON, and Req plus Jason handles it in about forty lines. You do not need an SDK. What you need is a correct picture of the wire format, because Claude’s request shape differs from OpenAI’s in a few places that will cost you an afternoon if you guess.

This guide covers the whole surface you actually use in a Phoenix app: a plain request, streaming server-sent events into a LiveView, the tool-use loop, structured JSON output, prompt caching, and the error handling that keeps a production feature from falling over. Every code block is real Elixir against the current API.

Why raw HTTP is the right call here

There are community options. Elixir LangChain has an Anthropic chat model and is a genuinely good fit if you want chains, tool orchestration and provider swapping out of the box. There are also thin wrapper packages on Hex.

But for a single provider and a handful of calls, wrapping Req yourself wins on three counts:

• No lag behind the API. New parameters (effort levels, structured outputs, caching TTLs) are available the day they ship, because you are writing the JSON.
• No hidden behaviour. You can see exactly what goes over the wire, which matters a lot when you are debugging a 400 at 11pm.
• Fewer dependencies. Req is already in most Phoenix apps.

The tradeoff is honest: you own the retry logic, the SSE parsing and the request shapes. That is maybe 150 lines of code, and this article is most of it.

Setup: Req, config, and the headers that matter

Add Req to mix.exs:

defp deps do
  [
    {:req, "~> 0.5"},
    {:jason, "~> 1.4"}
  ]
end

Keep the key in runtime config so it is read from the environment at boot, not baked into a release:

# config/runtime.exs
config :my_app, :anthropic,
  api_key: System.get_env("ANTHROPIC_API_KEY") ||
    raise("ANTHROPIC_API_KEY is not set")

Now the client module. Three headers are required on every request, and the first one is where people coming from OpenAI trip:

defmodule MyApp.Claude do
  @endpoint "https://api.anthropic.com/v1/messages"
  @version "2023-06-01"

  defp headers do
    [
      {"x-api-key", api_key()},
      {"anthropic-version", @version},
      {"content-type", "application/json"}
    ]
  end

  defp api_key, do: Application.fetch_env!(:my_app, :anthropic)[:api_key]
end

Note x-api-key, not Authorization: Bearer. The anthropic-version header is not optional, and pinning it is what stops a future API change from breaking your app silently.

Your first request

defmodule MyApp.Claude do
  # ... headers/0 and api_key/0 from above

  @doc "Sends a single message and returns the assistant's text."
  def ask(prompt, opts \\ []) do
    body = %{
      model: Keyword.get(opts, :model, "claude-opus-5"),
      max_tokens: Keyword.get(opts, :max_tokens, 16_000),
      system: Keyword.get(opts, :system),
      messages: [%{role: "user", content: prompt}]
    }

    case Req.post(@endpoint,
           json: body,
           headers: headers(),
           receive_timeout: 120_000
         ) do
      {:ok, %{status: 200, body: body}} -> {:ok, text_of(body)}
      {:ok, %{status: status, body: body}} -> {:error, {status, body}}
      {:error, reason} -> {:error, reason}
    end
  end

  # The response `content` is a list of blocks, not a string.
  defp text_of(%{"content" => blocks}) do
    blocks
    |> Enum.filter(&(&1["type"] == "text"))
    |> Enum.map_join("", & &1["text"])
  end
end

MyApp.Claude.ask("Summarise this changelog in three bullets: ...") and you are done.

Four differences from the OpenAI shape

If you have written the OpenAI version of this before (we have a full walkthrough of streaming OpenAI responses in Phoenix LiveView), these are the things that will bite:

1. max_tokens is required. There is no default. Omit it and you get a 400.
2. The system prompt is a top-level system field, not a message with role: "system" in the array.
3. content is a list of typed blocks on the way out, and can be either a plain string or a list of blocks on the way in. Text, tool calls and images are all blocks. Always iterate and filter by type rather than reaching for content[0].
4. stop_reason tells you what happened: "end_turn", "max_tokens", "tool_use", or "refusal". Branch on it before you read the content, especially inside a loop.

Which model to use

The current model IDs are claude-opus-5, claude-sonnet-5 and claude-haiku-4-5. Use the exact strings, with no date suffix appended. Opus is the default choice for quality-sensitive work, Sonnet is the mid tier, and Haiku is the cheap fast one for classification and routing. Input and output tokens bill at different rates, with output roughly five times the input rate across the range, so an endpoint that returns long text costs far more than one that reads long documents. Check Anthropic’s pricing page for current numbers before you model your unit economics.

One parameter worth knowing about early: output_config.effort, which takes low, medium, high, xhigh or max and controls how much reasoning the model spends on a request. Dropping a high-volume classification route to low is usually the largest cost lever you have that does not involve changing models:

body = %{
  model: "claude-opus-5",
  max_tokens: 1024,
  output_config: %{effort: "low"},
  messages: [%{role: "user", content: prompt}]
}

Streaming into a LiveView

This is where Elixir stops being the awkward choice and becomes the obvious one. Your page is already a stateful process behind a WebSocket, so streaming tokens is just message passing. No client-side state machine, no separate SSE endpoint, no JSON API in the middle.

Set stream: true and the response body arrives as server-sent events. Claude’s event stream is more structured than OpenAI’s: you get message_start, then content_block_start / content_block_delta / content_block_stop for each block, then message_delta carrying the final stop_reason and output token count, then message_stop.

The detail that every naive SSE parser gets wrong is that a TCP chunk does not respect event boundaries. You can and will receive half a JSON payload. So buffer the remainder between chunks. Req hands you the response struct in the accumulator, and Req.Response‘s private map is a clean place to keep that buffer:

defmodule MyApp.Claude do
  @doc """
  Streams a completion, sending `{:claude, {:chunk, text}}`,
  `{:claude, {:done, stop_reason}}` and `{:claude, {:error, term}}`
  to `pid`.
  """
  def stream(messages, pid, opts \\ []) do
    body = %{
      model: Keyword.get(opts, :model, "claude-opus-5"),
      max_tokens: Keyword.get(opts, :max_tokens, 8_000),
      stream: true,
      messages: messages
    }

    result =
      Req.post(@endpoint,
        json: body,
        headers: headers(),
        receive_timeout: 300_000,
        into: fn {:data, data}, {req, resp} ->
          buffer = Req.Response.get_private(resp, :sse_buffer, "")
          {events, rest} = split_events(buffer <> data)
          Enum.each(events, &handle_event(&1, pid))
          {:cont, {req, Req.Response.put_private(resp, :sse_buffer, rest)}}
        end
      )

    case result do
      {:ok, _resp} -> :ok
      {:error, reason} -> send(pid, {:claude, {:error, reason}})
    end
  end

  # SSE events are separated by a blank line. The tail may be a partial event.
  defp split_events(data) do
    parts = String.split(data, "\n\n")
    {complete, [rest]} = Enum.split(parts, length(parts) - 1)
    {complete, rest}
  end

  defp handle_event(event, pid) do
    event
    |> String.split("\n")
    |> Enum.filter(&String.starts_with?(&1, "data: "))
    |> Enum.each(fn "data: " <> json ->
      case Jason.decode(json) do
        {:ok, %{"type" => "content_block_delta",
                "delta" => %{"type" => "text_delta", "text" => text}}} ->
          send(pid, {:claude, {:chunk, text}})

        {:ok, %{"type" => "message_delta",
                "delta" => %{"stop_reason" => reason}}} ->
          send(pid, {:claude, {:done, reason}})

        {:ok, %{"type" => "error", "error" => error}} ->
          send(pid, {:claude, {:error, error}})

        _ ->
          # message_start, content_block_start/stop, ping, thinking deltas
          :ok
      end
    end)
  end
end

That catch-all clause matters. You will also see thinking_delta blocks if you turn on summarised thinking, and periodic ping events to keep the connection alive. Ignore what you do not render rather than crashing on it.

The LiveView side

The rule that governs everything here: never block the LiveView process. A LiveView handles one message at a time, so if you sit in the HTTP stream inside handle_event/3, the page freezes for the entire generation. Run the call in a supervised task and let tokens arrive as messages.

defmodule MyAppWeb.AssistantLive do
  use MyAppWeb, :live_view

  @impl true
  def mount(_params, _session, socket) do
    {:ok, assign(socket, answer: "", streaming?: false)}
  end

  @impl true
  def handle_event("ask", %{"prompt" => prompt}, socket) do
    pid = self()
    messages = [%{role: "user", content: prompt}]

    Task.Supervisor.async_nolink(MyApp.TaskSupervisor, fn ->
      MyApp.Claude.stream(messages, pid)
    end)

    {:noreply, assign(socket, answer: "", streaming?: true)}
  end

  @impl true
  def handle_info({:claude, {:chunk, text}}, socket) do
    {:noreply, assign(socket, answer: socket.assigns.answer <> text)}
  end

  def handle_info({:claude, {:done, _stop_reason}}, socket) do
    {:noreply, assign(socket, streaming?: false)}
  end

  def handle_info({:claude, {:error, error}}, socket) do
    {:noreply,
     socket
     |> assign(streaming?: false)
     |> put_flash(:error, "The assistant is unavailable: #{inspect(error)}")}
  end

  # async_nolink sends these two; ignore them.
  def handle_info({ref, _result}, socket) when is_reference(ref) do
    Process.demonitor(ref, [:flush])
    {:noreply, socket}
  end

  def handle_info({:DOWN, _ref, :process, _pid, _reason}, socket) do
    {:noreply, socket}
  end
end

Add {Task.Supervisor, name: MyApp.TaskSupervisor} to your application’s supervision tree. Using async_nolink under a supervisor means a crashed API call takes down the task and not the user’s page, and when the user navigates away the LiveView dies and takes the streaming task with it.

Rendering is unremarkable, which is the point:

<div class="prose">{@answer}</div>
<span :if={@streaming?} class="animate-pulse">Thinking...</span>

Tool use: the loop you have to write yourself

Tool use (function calling) is a conversation, not a single call. You declare the tools, Claude replies with stop_reason: "tool_use" and one or more tool_use blocks, you execute them and send the results back as a new user message, and you repeat until stop_reason is "end_turn".

defmodule MyApp.Claude.Agent do
  alias MyApp.Claude

  @tools [
    %{
      name: "search_orders",
      description: "Search a customer's orders by email address.",
      input_schema: %{
        type: "object",
        properties: %{
          email: %{type: "string", description: "Customer email"},
          status: %{type: "string", enum: ["open", "shipped", "refunded"]}
        },
        required: ["email"],
        additionalProperties: false
      },
      strict: true
    }
  ]

  def run(messages, depth \\ 0)

  def run(_messages, depth) when depth > 8, do: {:error, :tool_loop_limit}

  def run(messages, depth) do
    request = %{
      model: "claude-opus-5",
      max_tokens: 8_000,
      tools: @tools,
      messages: messages
    }

    case Claude.request(request) do
      {:ok, %{"stop_reason" => "tool_use", "content" => content}} ->
        results = Enum.flat_map(content, &execute/1)

        messages
        |> Kernel.++([
          %{role: "assistant", content: content},
          %{role: "user", content: results}
        ])
        |> run(depth + 1)

      {:ok, response} ->
        {:ok, response}

      error ->
        error
    end
  end

  defp execute(%{"type" => "tool_use", "id" => id, "name" => name, "input" => input}) do
    [%{type: "tool_result", tool_use_id: id, content: run_tool(name, input)}]
  rescue
    error ->
      [
        %{
          type: "tool_result",
          tool_use_id: id,
          content: "Tool failed: #{Exception.message(error)}",
          is_error: true
        }
      ]
  end

  defp execute(_block), do: []

  defp run_tool("search_orders", %{"email" => email} = input) do
    email
    |> MyApp.Orders.search(status: input["status"])
    |> Jason.encode!()
  end
end

Four things that are easy to get wrong here:

• Append the whole content list back as the assistant message. If you extract just the text and drop the tool_use blocks, the follow-up request is invalid, because your tool_result has nothing to point at.
• Return every result in a single user message. Claude can request several tools at once. Splitting the results across multiple messages quietly teaches the model to stop making parallel calls.
• Report failures with is_error: true rather than dropping the block. Claude will usually recover and try a different approach, which is far better than a silent hang.
• Bound the recursion. A depth cap turns a pathological loop into an error instead of a bill.

strict: true on a tool definition (with additionalProperties: false and a required list) guarantees the arguments validate against your schema, which means you can pattern match on them instead of defensively checking every field.

If you want to expose those same tools to Claude Code, Claude Desktop or any other MCP client rather than driving the loop yourself, that is a different protocol built on the same idea, and we covered it in building an MCP server in Elixir.

Structured output: JSON you can trust

For extraction work, do not ask for JSON in the prompt and hope. Constrain the response with output_config.format and the model is held to your schema:

def extract_invoice(text) do
  schema = %{
    type: "object",
    properties: %{
      supplier: %{type: "string"},
      invoice_number: %{type: "string"},
      issued_on: %{type: "string", format: "date"},
      total_cents: %{type: "integer"},
      currency: %{type: "string", enum: ["GBP", "USD", "EUR"]},
      line_items: %{
        type: "array",
        items: %{
          type: "object",
          properties: %{
            description: %{type: "string"},
            amount_cents: %{type: "integer"}
          },
          required: ["description", "amount_cents"],
          additionalProperties: false
        }
      }
    },
    required: [
      "supplier",
      "invoice_number",
      "issued_on",
      "total_cents",
      "currency",
      "line_items"
    ],
    additionalProperties: false
  }

  request = %{
    model: "claude-opus-5",
    max_tokens: 4_000,
    output_config: %{format: %{type: "json_schema", schema: schema}},
    messages: [%{role: "user", content: "Extract the invoice:\n\n" <> text}]
  }

  with {:ok, response} <- MyApp.Claude.request(request),
       json when is_binary(json) <- text_of(response) do
    Jason.decode(json)
  end
end

Schema rules worth memorising, because breaking them returns a 400 rather than degrading gracefully: every object needs additionalProperties: false and a required list, recursive schemas are not supported, and numeric or string constraints such as minimum and maxLength are not enforced. Validate those in an Ecto changeset after decoding, which is where they belong anyway.

Two operational notes. A new schema pays a small one-time compilation cost on its first request and is then cached, so do not generate schemas dynamically per request. And if stop_reason comes back as "max_tokens", the JSON is truncated and will not parse, so raise max_tokens instead of trying to repair the string.

This is the same technique behind document pipelines like extracting structured data from PDFs with AI in Elixir, where the schema becomes the contract between the model and your Ecto changeset.

Or skip the wiring entirely: phx_ai ships the streaming LiveView chat, the request plumbing and the key handling already assembled, so you start from a working AI feature instead of an empty Req call.

Prompt caching: the cheapest optimisation available

If you send the same large prefix on every request (a long system prompt, a product catalogue, a document you are answering questions about, retrieved chunks from a knowledge base), cache it. You mark the last block of the stable prefix with cache_control, and later requests read it back at a large discount instead of paying full input price.

body = %{
  model: "claude-opus-5",
  max_tokens: 4_000,
  system: [
    %{
      type: "text",
      text: MyApp.Prompts.support_agent_instructions(),
      cache_control: %{type: "ephemeral"}
    }
  ],
  messages: [%{role: "user", content: question}]
}

The mechanism is a prefix match, and that is the part people get wrong. The request renders in the order tools, then system, then messages. A single changed byte anywhere in the prefix invalidates everything after it. The classic mistake is interpolating a timestamp or a request ID into the system prompt, which silently drops your hit rate to zero while everything still appears to work.

Two habits fix it. Put stable content first and volatile content after the last cache breakpoint. And verify rather than assume:

%{
  "usage" => %{
    "input_tokens" => input,
    "output_tokens" => output,
    "cache_creation_input_tokens" => created,
    "cache_read_input_tokens" => read
  }
} = response

:telemetry.execute(
  [:my_app, :claude, :usage],
  %{input: input, output: output, cache_created: created, cache_read: read},
  %{model: response["model"]}
)

If cache_read_input_tokens is zero across repeated requests, something in your prefix is changing. One caveat: the minimum cacheable prefix is model-dependent and runs into the thousands of tokens, so short prompts will not cache at all, and that is expected rather than a bug.

Emitting those usage numbers as telemetry from day one is worth the five minutes. Token spend is the cost of goods sold for an AI feature, and if you bill customers for it you will eventually need per-account attribution, which is the subject of our guide to usage-based billing in Elixir and Phoenix.

Errors, retries and timeouts

Three failure modes matter in production:

• 429 rate limited. Respect the retry-after response header rather than backing off blindly.
• 529 overloaded. Transient upstream capacity pressure. Retry with backoff.
• Timeouts. A long generation can easily exceed a default HTTP timeout. Set receive_timeout generously (five minutes for streaming), and prefer streaming for anything with a large max_tokens, because a streamed response cannot hit a request timeout mid-generation.

Req covers the retryable cases with a short configuration:

Req.post(@endpoint,
  json: body,
  headers: headers(),
  receive_timeout: 300_000,
  retry: :transient,
  max_retries: 3,
  retry_log_level: :warning
)

What you should not retry is a 400. That means a malformed request (a missing max_tokens, an invalid schema, a tool_result with no matching tool_use_id), and retrying it just burns latency to fail again. Match on the status and let it fail loudly in development.

Finally, guard the user-facing path. An AI feature that returns a 500 when the upstream API has a bad ten minutes is a worse product than one that degrades to “the assistant is unavailable, here is the FAQ.” Put the call behind a timeout in your context module and give the LiveView something sensible to render.

Wrapping up

The Claude API in Elixir is genuinely pleasant once you stop looking for an SDK. Req handles the transport, pattern matching handles the content blocks, and OTP handles the concurrency problem that other ecosystems solve with extra infrastructure. Streaming in particular is less code in Phoenix than anywhere else, because the persistent connection you need already exists.

The pieces to take away: pin anthropic-version and use x-api-key, buffer your SSE parsing across chunks, always append the full content list in a tool loop, constrain extraction with output_config.format rather than prompt instructions, and cache your stable prefix while checking cache_read_input_tokens to prove it works.

If you would rather start from something that already has the streaming chat, the request layer and the key management wired up, phx_ai is built for exactly this. And if you are shipping more than one AI product, the Builder Pass gives you lifetime access to every template in the catalogue, AI starters included, for less than a couple of days spent building the plumbing yourself.