Claudio

Hex.pm Documentation CI License

A modern, feature-complete Elixir client for the Anthropic API

Claudio provides a comprehensive, idiomatic Elixir interface for Claude AI models with support for streaming, tool calling, prompt caching, vision, and batch processing.

Why Claudio?

Features

See the CHANGELOG for what changed in each release.

Installation

Add claudio to your list of dependencies in mix.exs:

def deps do
  [
    {:claudio, "~> 0.7"}
  ]
end

Claudio requires Elixir 1.15+. CI tests OTP 25 through 28.

Then fetch dependencies:

mix deps.get

Quick Start

1. Get an API Key

Sign up for an Anthropic API key at console.anthropic.com

2. Set Your API Key

export ANTHROPIC_API_KEY="your-api-key-here"

3. Send Your First Message

# Create a client
client = Claudio.Client.new(%{
  token: System.get_env("ANTHROPIC_API_KEY")
})

# Use the Request builder (recommended)
alias Claudio.Messages.{Request, Response}

request =
  Request.new("claude-opus-5-5")
  |> Request.add_message(:user, "Explain quantum computing in simple terms")
  |> Request.set_max_tokens(1024)

{:ok, response} = Claudio.Messages.create(client, request)

# Extract the text
text = Response.get_text(response)
IO.puts(text)

Examples

Multi-Turn Conversation

alias Claudio.Messages.{Request, Response}

request =
  Request.new("claude-opus-5-5")
  |> Request.set_system("You are a helpful Python tutor")
  |> Request.add_message(:user, "How do I read a file in Python?")
  |> Request.add_message(:assistant, "You can use the open() function...")
  |> Request.add_message(:user, "What about writing to a file?")
  |> Request.set_max_tokens(500)

{:ok, response} = Claudio.Messages.create(client, request)
IO.puts(Response.get_text(response))

Streaming Responses

Perfect for chat interfaces or real-time applications:

alias Claudio.Messages.{Request, Stream}

request =
  Request.new("claude-opus-5-5")
  |> Request.add_message(:user, "Write a haiku about Elixir")
  |> Request.set_max_tokens(100)
  |> Request.enable_streaming()

{:ok, stream_response} = Claudio.Messages.create(client, request)

# Print text as it arrives and get the final Response (usage, stop_reason, tool calls)
{:ok, response} = Stream.to_response(stream_response, on_text: &IO.write/1)
response.usage.output_tokens

If you only want the text chunks:

stream_response
|> Stream.parse_events()
|> Stream.accumulate_text()
|> Enum.each(&IO.write/1)

A streaming body can be read once, and only by the process that called Claudio.Messages.create/2. Pick one consumer per response; use :on_text / :on_event rather than enumerating it twice.

Tool/Function Calling

Let Claude use your functions:

alias Claudio.{Tools, Messages.Request, Messages.Response}

# Define a weather tool
weather_tool = Tools.define_tool(
  "get_weather",
  "Get current weather for a location",
  %{
    type: "object",
    properties: %{
      location: %{type: "string", description: "City name"},
      unit: %{type: "string", enum: ["celsius", "fahrenheit"]}
    },
    required: ["location"]
  }
)

# Create request with tool
request =
  Request.new("claude-opus-5-5")
  |> Request.add_message(:user, "What's the weather in Tokyo?")
  |> Request.add_tool(weather_tool)
  |> Request.set_max_tokens(500)

{:ok, response} = Claudio.Messages.create(client, request)

# Check if Claude wants to use the tool
if Tools.has_tool_uses?(response) do
  # Execute your function for every tool call, collecting the results
  tool_results =
    for tool_use <- Tools.extract_tool_uses(response) do
      result = get_weather(tool_use.input["location"])
      Tools.create_tool_result(tool_use.id, Jason.encode!(result))
    end

  # Replay Claude's turn, then send all results back in one user turn
  followup =
    request
    |> Request.add_message(:assistant, Response.to_assistant_content(response))
    |> Request.add_message(:user, tool_results)

  {:ok, final_response} = Claudio.Messages.create(client, followup)
  IO.puts(Response.get_text(final_response))
end

defp get_weather(location) do
  # Your weather API implementation
  %{temp: 72, condition: "sunny", location: location}
end

Vision - Analyze Images

# From a file
image_data = File.read!("screenshot.png") |> Base.encode64()

request =
  Request.new("claude-opus-5-5")
  |> Request.add_message_with_image(
    :user,
    "What's in this image?",
    image_data,
    "image/png"
  )
  |> Request.set_max_tokens(500)

{:ok, response} = Claudio.Messages.create(client, request)
IO.puts(Response.get_text(response))

# Or from a URL
request =
  Request.new("claude-opus-5-5")
  |> Request.add_message_with_image_url(
    :user,
    "Describe this diagram",
    "https://example.com/diagram.jpg"
  )
  |> Request.set_max_tokens(500)

Prompt Caching - Save 90% on Costs

Cache large contexts like documentation or code:

large_codebase = File.read!("lib/my_app.ex")

request =
  Request.new("claude-opus-5-5")
  |> Request.set_system_with_cache("""
    You are a code reviewer. Here is the codebase:

    #{large_codebase}

    Review code changes carefully for bugs and style.
    """, ttl: "5m")
  |> Request.add_message(:user, "Review this function: def foo(x), do: x + 1")
  |> Request.set_max_tokens(1000)

{:ok, response} = Claudio.Messages.create(client, request)

# Check cache savings
IO.inspect(response.usage.cache_read_input_tokens, label: "Tokens from cache")
IO.inspect(response.usage.cache_creation_input_tokens, label: "Tokens cached")

Batch Processing

Process thousands of requests asynchronously:

alias Claudio.Batches

# Create a batch of analysis tasks
requests =
  Enum.map(1..1000, fn i ->
    %{
      custom_id: "review-#{i}",
      params: %{
        model: "claude-opus-5-5",
        max_tokens: 500,
        messages: [
          %{role: "user", content: "Analyze pull request ##{i}"}
        ]
      }
    }
  end)

# Submit batch (processes asynchronously)
{:ok, batch} = Batches.create(client, requests)
IO.puts("Batch created: #{batch["id"]}")

# Wait for completion with progress updates
{:ok, _completed} =
  Batches.wait_for_completion(client, batch["id"],
    poll_interval: 10,  # seconds between checks
    callback: fn status ->
      counts = status["request_counts"]
      IO.puts("Done: #{counts["succeeded"] + counts["errored"]}, processing: #{counts["processing"]}")
    end
  )

# Download results: a list of decoded (string-keyed) maps
{:ok, results} = Batches.get_results(client, batch["id"])

Enum.each(results, fn result ->
  case result["result"]["type"] do
    "succeeded" ->
      message = result["result"]["message"]
      IO.puts("#{result["custom_id"]}: Success")

    "errored" ->
      error = result["result"]["error"]
      IO.puts("#{result["custom_id"]}: Error - #{error["message"]}")
  end
end)

Files API

Upload files to Anthropic's storage and reference them from message content blocks. The Files API is GA — no beta header needed. (Clients that still send files-api-2025-04-14 get the old list pagination; see Claudio.Files.list/2.)

alias Claudio.Files
alias Claudio.Messages.Request

client = Claudio.Client.new(%{
  token: System.get_env("ANTHROPIC_API_KEY")
})

# Upload a PDF
{:ok, %{"id" => file_id}} =
  Files.upload(client, File.read!("contract.pdf"),
    content_type: "application/pdf",
    filename: "contract.pdf"
  )

# Reference it from a message (no extra builder needed — the document helper
# already accepts a file_id)
request =
  Request.new("claude-opus-5-5")
  |> Request.add_message_with_document(:user, "Summarise this contract.", file_id)
  |> Request.set_max_tokens(1024)

{:ok, response} = Claudio.Messages.create(client, request)

# Manage uploaded files
{:ok, %{"data" => files}} = Files.list(client, limit: 50)
{:ok, _meta}              = Files.get(client, file_id)
{:ok, bytes}              = Files.download(client, file_id)
{:ok, _}                  = Files.delete(client, file_id)

Autonomous Agents

Run complex tool-calling loops with Claudio.Agent:

alias Claudio.{Agent, Tools}
alias Claudio.Messages.{Request, Response}

# Define your tools and their implementation logic
weather_tool = Tools.define_tool("get_weather", "Get weather", %{
  "type" => "object",
  "properties" => %{"location" => %{"type" => "string"}},
  "required" => ["location"]
})

handlers = %{
  "get_weather" => fn %{"location" => loc} ->
    {:ok, "72°F and sunny in #{loc}"}
  end
}

request =
  Request.new("claude-opus-5-5")
  |> Request.add_message(:user, "What's the weather in SF?")
  |> Request.add_tool(weather_tool)
  |> Request.set_max_tokens(1024)

# Agent.run handles the multi-turn loop automatically
{:ok, final_response, history} = Agent.run(client, request, handlers)

IO.puts(Response.get_text(final_response))

Managed Agents (beta)

Server-hosted agents that run in Anthropic's sandbox. Claudio attaches the managed-agents-2026-04-01 beta per request.

alias Claudio.ManagedAgents.{Agents, Environments, Sessions}

{:ok, agent} = Agents.create(client, %{name: "researcher", model: "claude-opus-5-5",
                                       tools: [%{type: "agent_toolset_20260401"}]})
{:ok, env} = Environments.create(client, %{name: "default",
                                           config: %{type: "cloud", networking: %{type: "unrestricted"}}})
{:ok, session} = Sessions.create(client, %{agent: agent["id"], environment_id: env["id"]})

{:ok, _} = Sessions.send_events(client, session["id"], [
  %{type: "user.message", content: [%{type: "text", text: "List the files in the repo."}]}
])

{:ok, %{"data" => events}} = Sessions.list_events(client, session["id"])

# Every page, lazily:
Claudio.ManagedAgents.stream(fn opts -> Sessions.list(client, opts) end, statuses: ["idle"])
|> Enum.map(& &1["id"])

A typed event stream and a run loop for custom tools and confirmations are planned next.

MCP (Model Context Protocol)

Connect Claude to any MCP server:

alias Claudio.MCP.{ToolAdapter, ResultMapper}
alias Claudio.Messages.Request

# 1. List tools from an MCP server (e.g. using ex_mcp)
{:ok, mcp_tools} = ExMCP.list_tools(mcp_client)

# 2. Add MCP tools to a Claudio request
request =
  Request.new("claude-opus-5-5")
  |> Request.add_message(:user, "Use your tools to search for Elixir libraries")
  |> ToolAdapter.add_tools(mcp_tools, prefix: "my_server")

{:ok, response} = Claudio.Messages.create(client, request)

# 3. Map results back to MCP calls
mcp_calls = ResultMapper.claudio_to_mcp(response)

Agent-to-Agent (A2A) Protocol

Communicate with other agents over a standardized protocol:

alias Claudio.A2A.{Client, Message, Part}

# Discover a remote agent's capabilities
{:ok, card} = Client.discover("https://expert-agent.com")

# Send a message over HTTP (the gRPC transport is not implemented)
message = Message.new(:user, [Part.text("Analyze this dataset")])
{:ok, task} = Client.send_message("https://expert-agent.com/a2a", message)

# Poll for task completion
{:ok, updated_task} = Client.get_task("https://expert-agent.com/a2a", task.id)

Telemetry & Monitoring

Claudio emits :telemetry events under five prefixes: [:claudio, :messages, :create], [:claudio, :messages, :count_tokens], [:claudio, :messages, :stream], [:claudio, :messages, :stream, :usage] and [:claudio, :http, :request].

defmodule MyApp.ClaudioTelemetry do
  require Logger

  def handle([:claudio, :messages, :create, :stop], measurements, metadata, _config) do
    ms = System.convert_time_unit(measurements.duration, :native, :millisecond)
    Logger.info("#{metadata.model} -> #{metadata[:response_model]} #{metadata.status} in #{ms}ms")
  end
end

:telemetry.attach("claudio-monitoring", [:claudio, :messages, :create, :stop],
  &MyApp.ClaudioTelemetry.handle/4, nil)

See the telemetry guide for every event, metadata key and an OpenTelemetry example.

Configuration

Basic Setup

# config/config.exs
config :claudio,
  default_api_version: "2023-06-01",
  default_beta_features: []

Per-Client Options

Timeouts and retries can be set on each client. A key given to Client.new/2 wins over the app config below, so one application can run differently-configured clients:

# Polling a batch: safe to retry
poller = Claudio.Client.new(%{token: key, recv_timeout: 600_000, retry: true})

# Creating a batch: a retried POST could create it twice
creator = Claudio.Client.new(%{token: key, retry: false})

Timeout Configuration (app-wide defaults)

Keys not passed to Client.new/2 fall back to the application config:

# config/config.exs
config :claudio, Claudio.Client,
  timeout: 60_000,        # Connection timeout: 60s
  recv_timeout: 120_000   # Receive timeout: 120s (important for streaming)

# For long-running operations
config :claudio, Claudio.Client,
  timeout: 60_000,
  recv_timeout: 600_000   # 10 minutes

# Production with retries
config :claudio, Claudio.Client,
  timeout: 30_000,
  recv_timeout: 180_000,
  retry: true  # Automatic retry on transient failures

Custom Retry Logic

config :claudio, Claudio.Client,
  retry: [
    delay: 1000,          # Initial delay: 1s
    max_retries: 3,       # Retry up to 3 times
    max_delay: 10_000     # Max delay: 10s
  ]

The same values work per client (Client.new(%{token: key, retry: [max_retries: 5]})). Any other retry value, or an unknown key, raises ArgumentError.

Error Handling

Claudio provides structured error types for pattern matching:

alias Claudio.APIError

case Claudio.Messages.create(client, request) do
  {:ok, response} ->
    # Success
    handle_response(response)

  {:error, %APIError{type: :rate_limit_error} = error} ->
    # Rate limited - wait and retry
    Logger.warning("Rate limited: #{error.message}")
    Process.sleep(60_000)
    retry_request()

  {:error, %APIError{type: :authentication_error}} ->
    # Invalid API key
    Logger.error("Authentication failed - check your API key")

  {:error, %APIError{type: :invalid_request_error} = error} ->
    # Bad request - fix and retry
    Logger.error("Invalid request: #{error.message}")
    fix_and_retry()

  {:error, %APIError{type: :overloaded_error}} ->
    # Service overloaded - retry with backoff
    exponential_backoff_retry()

  {:error, %APIError{} = error} ->
    # Other API error
    Logger.error("API error [#{error.status_code}]: #{error.message}")

  {:error, reason} ->
    # Network or timeout error
    Logger.error("Request failed: #{inspect(reason)}")
end

Error Types

Best Practices

1. Use the Request Builder

The fluent Request API is more maintainable than raw maps:

# Good ✓
request =
  Request.new("claude-opus-5-5")
  |> Request.add_message(:user, "Hello")
  |> Request.set_max_tokens(100)

# Works, but less maintainable
request = %{
  "model" => "claude-opus-5-5",
  "messages" => [%{"role" => "user", "content" => "Hello"}],
  "max_tokens" => 100
}

2. Handle Errors Properly

Always pattern match on error types:

# Good ✓
case Claudio.Messages.create(client, request) do
  {:ok, response} -> handle_success(response)
  {:error, %APIError{type: :rate_limit_error}} -> retry_with_backoff()
  {:error, error} -> handle_error(error)
end

# Bad ✗
{:ok, response} = Claudio.Messages.create(client, request)  # Crashes on error

3. Use System Prompts

Guide the model's behavior with system prompts:

request =
  Request.new("claude-opus-5-5")
  |> Request.set_system("You are a helpful coding assistant. Always explain your code.")
  |> Request.add_message(:user, "Write a function to reverse a string")

4. Set Appropriate Timeouts

Long operations need longer timeouts:

# For batch processing or large responses
config :claudio, Claudio.Client,
  recv_timeout: 600_000  # 10 minutes

5. Enable Retries in Production

Handle transient failures automatically:

config :claudio, Claudio.Client,
  retry: true

6. Cache Large Contexts

Use prompt caching for repeated contexts:

# Cache documentation or code for multiple queries
request =
  Request.new("claude-opus-5-5")
  |> Request.set_system_with_cache(large_documentation, ttl: "5m")

7. Count Tokens for Cost Control

{:ok, count} = Claudio.Messages.count_tokens(client, request)
estimated_cost = count["input_tokens"] * 0.003 / 1000
IO.puts("Estimated cost: $#{estimated_cost}")

Testing

# Run unit tests
mix test

# Run with integration tests (requires ANTHROPIC_API_KEY)
export ANTHROPIC_API_KEY="your-key"
mix test --include integration

# Run specific test file
mix test test/messages_test.exs

# Check code formatting
mix format --check-formatted

# Everything CI checks locally: compile warnings, unused deps, format, Credo, Dialyzer, tests
mix precommit

Documentation

Full API documentation is available on HexDocs:

Generate documentation locally:

mix docs
open doc/index.html

Contributing

Contributions are welcome! Please feel free to submit a Pull Request.

  1. Fork the repository
  2. Create your feature branch (git checkout -b feature/amazing-feature)
  3. Write tests for your changes
  4. Ensure all tests pass (mix test)
  5. Commit your changes (git commit -am 'Add amazing feature')
  6. Push to the branch (git push origin feature/amazing-feature)
  7. Open a Pull Request

License

This project is licensed under the MIT License - see the LICENSE file for details.

Links

Acknowledgments

Built with ❤️ using Req for HTTP client operations.

Made with Elixir 💜