# `Anubis.Client` MCP (Model Context Protocol) client for connecting to MCP servers. This module provides a fully functional MCP client with automatic supervision, transport management, and all standard MCP operations. No macros needed — just add it to your supervision tree with the desired configuration. ## Usage Add the client to your supervision tree: children = [ {Anubis.Client, name: MyApp.MCPClient, transport: {:stdio, command: "uvx", args: ["mcp-server-anthropic"]}, client_info: %{"name" => "MyApp", "version" => "1.0.0"}, capabilities: %{"roots" => %{}}, protocol_version: "2025-06-18"} ] Use the client by passing the registered name: {:ok, tools} = Anubis.Client.list_tools(MyApp.MCPClient) {:ok, result} = Anubis.Client.call_tool(MyApp.MCPClient, "search", %{query: "elixir"}) ## Capabilities Capabilities are passed as a map with string keys: %{"roots" => %{}, "sampling" => %{}} For convenience, use `parse_capability/2` to build from atoms: capabilities = [:roots, {:sampling, list_changed?: true}] |> Enum.reduce(%{}, &Anubis.Client.parse_capability/2) ## Transport Configuration When starting the client, provide transport configuration: * `{:stdio, command: "cmd", args: ["arg1", "arg2"]}` * `{:sse, base_url: "http://localhost:8000"}` * `{:websocket, url: "ws://localhost:8000/ws"}` * `{:streamable_http, url: "http://localhost:8000/mcp"}` ## Process Naming The `:name` option controls process registration. You can use any valid `GenServer.name()` — an atom, a PID, or a `{:via, module, term}` tuple: # Atom name {Anubis.Client, name: MyApp.MCPClient, transport: ...} # For distributed systems with registries (e.g., Horde) {Anubis.Client, name: {:via, Horde.Registry, {MyCluster, "client_1"}}, transport_name: {:via, Horde.Registry, {MyCluster, "transport_1"}}, transport: ...} When using via tuples or other non-atom names, you must explicitly provide the `:transport_name` option. For atom names, the transport is automatically named as `Module.concat(ClientName, "Transport")`. ## Dynamic Client Management For applications that need to manage multiple client connections dynamically (e.g., user-configured MCP servers), use a `DynamicSupervisor`: DynamicSupervisor.start_child( MyApp.DynamicSupervisor, {Anubis.Client, name: {:via, Registry, {MyApp.Registry, client_id}}, transport_name: {:via, Registry, {MyApp.Registry, {client_id, :transport}}}, transport: {:streamable_http, base_url: url}, client_info: %{"name" => "MyApp", "version" => "1.0.0"}, capabilities: %{}, protocol_version: "2025-06-18"} ) # `capabilities` ```elixir @type capabilities() :: %{ optional(:roots | String.t()) => %{ optional(:listChanged | String.t()) => boolean() }, optional(:sampling | String.t()) => %{} } ``` MCP client capabilities - `:roots` - Capabilities related to the roots resource - `:listChanged` - Whether the client can handle listChanged notifications - `:sampling` - Capabilities related to sampling MCP describes these client capabilities on it [specification](https://spec.modelcontextprotocol.io/specification/2024-11-05/client/) # `capabilities_input` ```elixir @type capabilities_input() :: [ capability() | {capability(), capability_opts()} | map() ] ``` # `capability` ```elixir @type capability() :: :roots | :sampling ``` # `capability_opts` ```elixir @type capability_opts() :: [{:list_changed?, boolean()}] ``` # `client_info` ```elixir @type client_info() :: %{ required(:name | String.t()) => String.t(), optional(:version | String.t()) => String.t() } ``` MCP client metadata info - `:name` - The name of the client (required) - `:version` - The version of the client # `log_callback` ```elixir @type log_callback() :: (level :: String.t(), data :: term(), logger :: String.t() | nil -> any()) ``` Log callback function type. Called when log message notifications are received from the server. ## Parameters - `level` - Log level as a string (e.g., "debug", "info", "warning", "error") - `data` - Log message data, typically a map with message details - `logger` - Optional logger name identifying the source ## Returns - The return value is ignored # `option` ```elixir @type option() :: {:name, GenServer.name()} | {:transport, transport()} | {:client_info, map()} | {:capabilities, map()} | {:protocol_version, String.t()} | GenServer.option() ``` MCP client initialization options - `:name` - Following the `GenServer` patterns described on "Name registration". - `:transport` - The MCP transport options - `:client_info` - Information about the client - `:capabilities` - Client capabilities to advertise to the MCP server - `:protocol_version` - Protocol version to use (defaults to "2024-11-05") Any other option support by `GenServer`. # `progress_callback` ```elixir @type progress_callback() :: (progress_token :: String.t() | integer(), progress :: number(), total :: number() | nil -> any()) ``` Progress callback function type. Called when progress notifications are received for a specific progress token. ## Parameters - `progress_token` - String or integer identifier for the progress operation - `progress` - Current progress value - `total` - Total expected value (nil if unknown) ## Returns - The return value is ignored # `root` ```elixir @type root() :: %{uri: String.t(), name: String.t() | nil} ``` Root directory specification. Represents a root directory that the client has access to. ## Fields - `:uri` - File URI for the root directory (e.g., "file:///home/user/project") - `:name` - Optional human-readable name for the root # `t` ```elixir @type t() :: GenServer.server() ``` # `transport` ```elixir @type transport() :: [ layer: Anubis.Transport.STDIO | Anubis.Transport.SSE | Anubis.Transport.WebSocket | Anubis.Transport.StreamableHTTP, name: GenServer.server() ] ``` MCP client transport options - `:layer` - The transport layer to use, either `Anubis.Transport.STDIO`, `Anubis.Transport.SSE`, `Anubis.Transport.WebSocket`, or `Anubis.Transport.StreamableHTTP` (required) - `:name` - The transport optional custom name # `add_root` ```elixir @spec add_root(t(), String.t(), String.t() | nil, opts :: Keyword.t()) :: :ok ``` Adds a root directory to the client's roots list. ## Parameters * `client` - The client process * `uri` - The URI of the root directory (must start with "file://") * `name` - Optional human-readable name for the root * `opts` - Additional options * `:timeout` - Request timeout in milliseconds # `await_ready` ```elixir @spec await_ready( t(), keyword() ) :: :ok ``` Blocks until the client has completed the MCP initialization handshake. Returns `:ok` once the server capabilities have been received. If the server has already been initialized, returns immediately. Otherwise, the caller is parked until the initialization response arrives or the GenServer call times out. ## Options * `:timeout` - Maximum time to wait in milliseconds (default: 30s) ## Examples {:ok, _supervisor} = Anubis.Client.start_link(opts) :ok = Anubis.Client.await_ready(MyApp.MCPClient, timeout: 10_000) {:ok, tools} = Anubis.Client.list_tools(MyApp.MCPClient) # `call_tool` ```elixir @spec call_tool(t(), String.t(), map() | nil, keyword()) :: {:ok, Anubis.MCP.Response.t()} | {:error, Anubis.MCP.Error.t()} ``` Calls a tool on the server. ## Options * `:timeout` - Request timeout in milliseconds * `:progress` - Progress tracking options * `:token` - A unique token to track progress (string or integer) * `:callback` - A function to call when progress updates are received # `cancel_all_requests` ```elixir @spec cancel_all_requests(t(), String.t(), opts :: Keyword.t()) :: {:ok, [Anubis.Client.Request.t()]} | {:error, Anubis.MCP.Error.t()} ``` Cancels all pending requests. ## Parameters * `client` - The client process * `reason` - Optional reason for cancellation (defaults to "client_cancelled") ## Returns * `{:ok, requests}` - A list of the Request structs that were cancelled * `{:error, reason}` - If an error occurred # `cancel_request` ```elixir @spec cancel_request(t(), String.t(), String.t(), opts :: Keyword.t()) :: :ok | {:error, Anubis.MCP.Error.t()} ``` Cancels an in-progress request. ## Parameters * `client` - The client process * `request_id` - The ID of the request to cancel * `reason` - Optional reason for cancellation ## Returns * `:ok` if the cancellation was successful * `{:error, reason}` if an error occurred * `{:not_found, request_id}` if the request ID was not found # `child_spec` Returns a child specification for starting the client under a supervisor. This starts a supervision tree containing both the client GenServer and the configured transport process, linked with a `:one_for_all` strategy. # `clear_roots` ```elixir @spec clear_roots(t(), opts :: Keyword.t()) :: :ok ``` Clears all root directories. ## Parameters * `client` - The client process * `opts` - Additional options * `:timeout` - Request timeout in milliseconds # `close` ```elixir @spec close(t()) :: :ok ``` Closes the client connection and terminates the process. # `complete` ```elixir @spec complete(t(), map(), map(), keyword()) :: {:ok, Anubis.MCP.Response.t()} | {:error, Anubis.MCP.Error.t()} ``` Requests autocompletion suggestions for prompt arguments or resource URIs. ## Parameters * `client` - The client process * `ref` - Reference to what is being completed (required) * For prompts: `%{"type" => "ref/prompt", "name" => prompt_name}` * For resources: `%{"type" => "ref/resource", "uri" => resource_uri}` * `argument` - The argument being completed (required) * `%{"name" => arg_name, "value" => current_value}` * `opts` - Additional options * `:timeout` - Request timeout in milliseconds * `:progress` - Progress tracking options * `:token` - A unique token to track progress (string or integer) * `:callback` - A function to call when progress updates are received ## Returns Returns `{:ok, response}` with completion suggestions if successful, or `{:error, reason}` if an error occurs. The response result contains a "completion" object with: * `values` - List of completion suggestions (maximum 100) * `total` - Optional total number of matching items * `hasMore` - Boolean indicating if more results are available # `get_prompt` ```elixir @spec get_prompt(t(), String.t(), map() | nil, keyword()) :: {:ok, Anubis.MCP.Response.t()} | {:error, Anubis.MCP.Error.t()} ``` Gets a specific prompt from the server. ## Options * `:timeout` - Request timeout in milliseconds * `:progress` - Progress tracking options * `:token` - A unique token to track progress (string or integer) * `:callback` - A function to call when progress updates are received # `get_schema` # `get_server_capabilities` ```elixir @spec get_server_capabilities(t(), opts :: Keyword.t()) :: map() | nil ``` Gets the server's capabilities as reported during initialization. Returns `nil` if the client has not been initialized yet. # `get_server_info` ```elixir @spec get_server_info(t(), opts :: Keyword.t()) :: map() | nil ``` Gets the server's information as reported during initialization. Returns `nil` if the client has not been initialized yet. # `is_client_capability` *macro* Guard to check if an atom is a valid client capability. # `is_supported_capability` *macro* Guard to check if a capability is supported by checking map keys. # `list_prompts` ```elixir @spec list_prompts( t(), keyword() ) :: {:ok, Anubis.MCP.Response.t()} | {:error, Anubis.MCP.Error.t()} ``` Lists available prompts from the server. ## Options * `:cursor` - Pagination cursor for continuing a previous request * `:timeout` - Request timeout in milliseconds * `:progress` - Progress tracking options * `:token` - A unique token to track progress (string or integer) * `:callback` - A function to call when progress updates are received # `list_resource_templates` ```elixir @spec list_resource_templates( t(), keyword() ) :: {:ok, Anubis.MCP.Response.t()} | {:error, Anubis.MCP.Error.t()} ``` Lists available resource templates from the server. ## Options * `:cursor` - Pagination cursor for continuing a previous request * `:timeout` - Request timeout in milliseconds * `:progress` - Progress tracking options * `:token` - A unique token to track progress (string or integer) * `:callback` - A function to call when progress updates are received # `list_resources` ```elixir @spec list_resources( t(), keyword() ) :: {:ok, Anubis.MCP.Response.t()} | {:error, Anubis.MCP.Error.t()} ``` Lists available resources from the server. ## Options * `:cursor` - Pagination cursor for continuing a previous request * `:timeout` - Request timeout in milliseconds * `:progress` - Progress tracking options * `:token` - A unique token to track progress (string or integer) * `:callback` - A function to call when progress updates are received # `list_roots` ```elixir @spec list_roots(t(), opts :: Keyword.t()) :: [map()] ``` Gets a list of all root directories. ## Parameters * `client` - The client process * `opts` - Additional options * `:timeout` - Request timeout in milliseconds # `list_tools` ```elixir @spec list_tools( t(), keyword() ) :: {:ok, Anubis.MCP.Response.t()} | {:error, Anubis.MCP.Error.t()} ``` Lists available tools from the server. ## Options * `:cursor` - Pagination cursor for continuing a previous request * `:timeout` - Request timeout in milliseconds * `:progress` - Progress tracking options * `:token` - A unique token to track progress (string or integer) * `:callback` - A function to call when progress updates are received # `merge_capabilities` ```elixir @spec merge_capabilities(t(), map(), opts :: Keyword.t()) :: map() ``` Merges additional capabilities into the client's capabilities. # `parse_capability` ```elixir @spec parse_capability(capability() | {capability(), capability_opts()}, map()) :: map() ``` Converts a capability atom or tuple into a map entry. Useful for building capability maps from ergonomic shorthand: capabilities = [:roots, {:sampling, list_changed?: true}] |> Enum.reduce(%{}, &Anubis.Client.parse_capability/2) # => %{"roots" => %{}, "sampling" => %{}} # `parse_options` # `parse_options!` # `ping` ```elixir @spec ping( t(), keyword() ) :: :pong | {:error, Anubis.MCP.Error.t()} ``` Sends a ping request to the server to check connection health. Returns `:pong` if successful. ## Options * `:timeout` - Request timeout in milliseconds (default: 30s) * `:progress` - Progress tracking options * `:token` - A unique token to track progress (string or integer) * `:callback` - A function to call when progress updates are received # `read_resource` ```elixir @spec read_resource(t(), String.t(), keyword()) :: {:ok, Anubis.MCP.Response.t()} | {:error, Anubis.MCP.Error.t()} ``` Reads a specific resource from the server. ## Options * `:timeout` - Request timeout in milliseconds * `:progress` - Progress tracking options * `:token` - A unique token to track progress (string or integer) * `:callback` - A function to call when progress updates are received # `register_log_callback` ```elixir @spec register_log_callback(t(), log_callback(), opts :: Keyword.t()) :: :ok ``` Registers a callback function to be called when log messages are received. ## Parameters * `client` - The client process * `callback` - A function that takes three arguments: level, data, and logger name The callback function will be called whenever a log message notification is received. # `register_progress_callback` ```elixir @spec register_progress_callback( t(), String.t() | integer(), progress_callback(), opts :: Keyword.t() ) :: :ok ``` Registers a callback function to be called when progress notifications are received for the specified progress token. ## Parameters * `client` - The client process * `progress_token` - The progress token to watch for (string or integer) * `callback` - A function that takes three arguments: progress_token, progress, and total The callback function will be called whenever a progress notification with the matching token is received. # `register_sampling_callback` ```elixir @spec register_sampling_callback( t(), (map() -> {:ok, map()} | {:error, String.t()}) ) :: :ok ``` Registers a callback function to handle sampling requests from the server. The callback function will be called when the server sends a `sampling/createMessage` request. The callback should implement user approval and return the LLM response. ## Callback Function The callback receives the sampling parameters and must return: - `{:ok, response_map}` - Where response_map contains: - `"role"` - Usually "assistant" - `"content"` - Message content (text, image, or audio) - `"model"` - The model that was used - `"stopReason"` - Why generation stopped (e.g., "endTurn") - `{:error, reason}` - If the user rejects or an error occurs # `remove_root` ```elixir @spec remove_root(t(), String.t(), opts :: Keyword.t()) :: :ok ``` Removes a root directory from the client's roots list. ## Parameters * `client` - The client process * `uri` - The URI of the root directory to remove * `opts` - Additional options * `:timeout` - Request timeout in milliseconds # `send_progress` ```elixir @spec send_progress( t(), String.t() | integer(), number(), number() | nil, opts :: Keyword.t() ) :: :ok | {:error, term()} ``` Sends a progress notification to the server for a long-running operation. ## Parameters * `client` - The client process * `progress_token` - The progress token provided in the original request (string or integer) * `progress` - The current progress value (number) * `total` - The optional total value for the operation (number) Returns `:ok` if notification was sent successfully, or `{:error, reason}` otherwise. # `set_log_level` ```elixir @spec set_log_level(t(), String.t()) :: {:ok, Anubis.MCP.Response.t()} | {:error, Anubis.MCP.Error.t()} ``` Sets the minimum log level for the server to send log messages. ## Parameters * `client` - The client process * `level` - The minimum log level (debug, info, notice, warning, error, critical, alert, emergency) Returns {:ok, result} if successful, {:error, reason} otherwise. # `start_link` ```elixir @spec start_link(keyword()) :: Supervisor.on_start() ``` Starts the client supervision tree (client + transport). This is the primary entry point for starting a client. It creates a supervisor that manages both the client GenServer and the transport process. # `unregister_log_callback` ```elixir @spec unregister_log_callback(t(), opts :: Keyword.t()) :: :ok ``` Unregisters a previously registered log callback. ## Parameters * `client` - The client process * `callback` - The callback function to unregister # `unregister_progress_callback` ```elixir @spec unregister_progress_callback(t(), String.t() | integer(), opts :: Keyword.t()) :: :ok ``` Unregisters a previously registered progress callback for the specified token. ## Parameters * `client` - The client process * `progress_token` - The progress token to stop watching (string or integer) # `unregister_sampling_callback` ```elixir @spec unregister_sampling_callback(t()) :: :ok ``` Unregisters the sampling callback. --- *Consult [api-reference.md](api-reference.md) for complete listing*