Building a Client

An MCP client connects your application to a server, whether that server is written in Elixir, Python, TypeScript, or anything else. In Anubis a client is a supervised process that owns one connection, negotiates the handshake, and exposes the server's capabilities as ordinary function calls.

Starting a client

Add Anubis.Client to your supervision tree:

children = [
  {Anubis.Client,
   name: MyApp.WeatherClient,
   transport: {:stdio, command: "weather-server", args: []},
   client_info: %{"name" => "MyApp", "version" => "1.0.0"},
   capabilities: %{}}
]

Supervisor.start_link(children, strategy: :one_for_one)

Four options matter here:

• name registers the process, and every client function takes this name (or a PID) as its first argument.
• transport says how to reach the server. See Transports for the full set.
• client_info identifies your application to the server during the handshake.
• capabilities declares what your client supports. An empty map is valid and common. This option is required.

The child spec starts a small supervision tree holding the client process and its transport, linked one_for_all, so a transport crash restarts both cleanly.

The handshake runs asynchronously after startup. If you need to block until the connection is ready, for example in a test or a script, use await_ready/2:

:ok = Anubis.Client.await_ready(MyApp.WeatherClient)

Connection eras

The MCP specification now has two eras. Legacy versions (up to 2025-11-25) open a connection with the initialize handshake and hold a session. The 2026-07-28 revision is stateless: there is no handshake and no session, and every request carries its protocol version, capabilities and client identity in params._meta.

Anubis clients detect the era themselves. On start the client probes the server with server/discover:

• a server that answers in the stateless era is used statelessly from then on: no initialize is sent, no session exists, and every request carries the era _meta with your configured capabilities and client info;
• every other probe outcome - a legacy-only server answering -32601, a plain HTTP 400/404 body, a timeout - falls back to today's initialize handshake, byte for byte.

await_ready/2 returns :ok in both eras once detection or the handshake completes. Ask which era the client settled on with get_era/2:

Anubis.Client.get_era(MyApp.WeatherClient)
#=> :stateless

The values are :detecting (nothing sent yet), :connecting (a probe or an initialize is in flight), :legacy, and :stateless. One client process is one connection to one server, so the era is cached in the client process itself.

A few things change when the era is :stateless:

• ping/2, set_log_level/2, subscribe_resource/3 and unsubscribe_resource/3 are refused with a method_not_found error naming the era: the 2026-07-28 dialect removed them. register_log_callback/3 keeps working (notifications/message still exists), and roots keep working on local state.
• The server's identity and capabilities come from the discover result, so get_server_info/2 and get_server_capabilities/2 behave as in the legacy era.
• A cached stateless version can be rejected later by a server config change or mixed pods behind a load balancer. The client re-negotiates from the error's version list and re-sends the request once on the same id; the request was never executed, so this is safe. A second rejection surfaces the error.

The :protocol_version option keeps its shape and changes its meaning by era:

• a legacy version (the default) stays the initialize proposal, as before;
• a registered stateless version (now accepted at start) becomes the preferred probe version, and the fallback handshake proposes the newest legacy version instead, with a warning.

Most applications can ignore all of this: detection is automatic and the fallback keeps legacy servers working unchanged.

Discovering capabilities

Once connected you can ask the server what it offers:

info = Anubis.Client.get_server_info(MyApp.WeatherClient)
caps = Anubis.Client.get_server_capabilities(MyApp.WeatherClient)

{:ok, %{result: %{"tools" => tools}}} = Anubis.Client.list_tools(MyApp.WeatherClient)

for tool <- tools do
  IO.puts("#{tool["name"]}: #{tool["description"]}")
end

list_resources/2, list_prompts/2, and list_resource_templates/2 follow the same shape. All list functions accept a cursor: option for pagination when the server paginates.

Calling tools

{:ok, response} =
  Anubis.Client.call_tool(MyApp.WeatherClient, "get_forecast", %{
    "location" => "Tokyo",
    "days" => 5
  })

Requests return {:ok, %Anubis.MCP.Response{}} or {:error, %Anubis.MCP.Error{}}. The response struct carries the raw result map and an is_error flag, which distinguishes the two failure levels MCP defines:

case Anubis.Client.call_tool(MyApp.WeatherClient, "get_weather", %{"location" => ""}) do
  {:ok, %{is_error: false, result: result}} ->
    handle_weather(result)

  {:ok, %{is_error: true, result: error}} ->
    Logger.warning("tool reported failure: #{inspect(error)}")

  {:error, %Anubis.MCP.Error{} = error} ->
    Logger.error("protocol or transport failure: #{inspect(error)}")
end

A tool-level error (is_error: true) means the server ran the tool and the tool failed; the payload usually explains why. A protocol error means the request itself did not complete: the server rejected it, the transport dropped, or the call timed out.

Reading resources and prompts

{:ok, %{result: %{"contents" => contents}}} =
  Anubis.Client.read_resource(MyApp.WeatherClient, "weather://stations/KSFO")

for content <- contents do
  case content do
    %{"text" => text} -> handle_text(text)
    %{"blob" => blob} -> handle_binary(blob)
  end
end

Prompts work the same way, with arguments:

{:ok, %{result: %{"messages" => messages}}} =
  Anubis.Client.get_prompt(MyApp.WeatherClient, "storm_briefing", %{"region" => "pacific"})

Servers that declare resource subscriptions also support subscribe_resource/3 and unsubscribe_resource/3.

Timeouts and progress

Every request function accepts a timeout: option in milliseconds, defaulting to 30 seconds:

Anubis.Client.call_tool(client, "analyze_dataset", params, timeout: to_timeout(minute: 5))

For long-running operations MCP defines progress notifications. Generate a token, pass it with the request, and optionally attach a callback:

token = Anubis.MCP.ID.generate_progress_token()

callback = fn ^token, progress, total ->
  IO.puts("progress: #{progress}/#{total || "?"}")
end

Anubis.Client.call_tool(client, "analyze_dataset", params,
  progress: [token: token, callback: callback]
)

The callback runs each time the server reports progress for that token.

Multiple connections

Each client owns exactly one connection, so connecting to several servers means starting several clients:

children = [
  {Anubis.Client,
   name: MyApp.SearchClient,
   transport: {:stdio, command: "search-server", args: []},
   client_info: %{"name" => "MyApp", "version" => "1.0.0"},
   capabilities: %{}},
  {Anubis.Client,
   name: MyApp.FilesClient,
   transport: {:streamable_http, base_url: "http://localhost:8000"},
   client_info: %{"name" => "MyApp", "version" => "1.0.0"},
   capabilities: %{}}
]

When connections are created at runtime, for example from user-configured server URLs, start clients under a DynamicSupervisor:

children = [
  {DynamicSupervisor, name: MyApp.MCPSupervisor, strategy: :one_for_one}
]

def connect(user_id, url) do
  spec =
    {Anubis.Client,
     name: {:via, Registry, {MyApp.MCPRegistry, user_id}},
     transport_name: {:via, Registry, {MyApp.MCPRegistry, {user_id, :transport}}},
     transport: {:streamable_http, base_url: url},
     client_info: %{"name" => "MyApp", "version" => "1.0.0"},
     capabilities: %{}}

  DynamicSupervisor.start_child(MyApp.MCPSupervisor, spec)
end

Atom names derive a transport name automatically. With :via names you must provide transport_name: yourself, as above. All client functions accept the name or the PID returned by start_child/2.

Client capabilities

Some MCP features flow from server to client: sampling asks your client to run an LLM completion, roots lets the server ask which directories it may touch, and elicitation requests structured input from the user. Declare the ones you support and register a handler:

{Anubis.Client,
 name: MyApp.MCPClient,
 transport: {:stdio, command: "server", args: []},
 client_info: %{"name" => "MyApp", "version" => "1.0.0"},
 capabilities: %{"sampling" => %{}, "roots" => %{}}}

Anubis.Client.register_sampling_callback(MyApp.MCPClient, fn request ->
  {:ok, MyApp.LLM.complete(request)}
end)

Anubis.Client.add_root(MyApp.MCPClient, "file:///home/user/project", "project")

Anubis.Client.parse_capability/2 builds the capability map from atom shorthand when you prefer:

capabilities =
  Enum.reduce([:sampling, {:roots, list_changed?: true}], %{}, &Anubis.Client.parse_capability/2)

list_changed?: true is only valid for :roots — it is the single client capability with a listChanged flag in the MCP schema. Passing it with :sampling or :elicitation raises ArgumentError.

Server logs

Servers with the :logging capability can push log messages to your client. You can set the minimum level and handle the stream:

Anubis.Client.set_log_level(MyApp.MCPClient, "warning")

Anubis.Client.register_log_callback(MyApp.MCPClient, fn level, data, logger ->
  Logger.log(String.to_existing_atom(level), "mcp[#{logger}]: #{inspect(data)}")
end)

Shutting down

Anubis.Client.close(MyApp.MCPClient)

This closes the connection and stops the transport. Under a supervisor you rarely call it yourself; stopping the supervisor tree does the same work.

Runnable web search example

The Parallel Search client uses Streamable HTTP to search the web and fetch page excerpts without an API key. It includes installation instructions and direct tool calls through Anubis.Client.

Next steps

• Transports details each transport option and when to pick it.
• Building a Server covers the other side of the connection.
• Recipes includes patterns for progress, logging, and error recovery.