Anubis.Server behaviour (anubis_mcp v0.14.0)

Build MCP servers that extend language model capabilities.

MCP servers are specialized processes that provide three core primitives to AI assistants: Resources (contextual data like files or schemas), Tools (actions the model can invoke), and Prompts (user-selectable templates). They operate in a secure, isolated architecture where clients maintain 1:1 connections with servers, enabling composable functionality while maintaining strict security boundaries.

Quick Start

Create a server in three steps:

defmodule MyServer do
  use Anubis.Server,
    name: "my-server",
    version: "1.0.0",
    capabilities: [:tools]

  component MyServer.Calculator
end

defmodule MyServer.Calculator do
  use Anubis.Server.Component, type: :tool

  def definition do
    %{
      name: "add",
      description: "Add two numbers",
      input_schema: %{
        type: "object",
        properties: %{
          a: %{type: "number"},
          b: %{type: "number"}
        }
      }
    }
  end

  def call(%{"a" => a, "b" => b}), do: {:ok, a + b}
end

# Start your server
{:ok, _pid} = Anubis.Server.start_link(MyServer, [], transport: :stdio)

Your server is now a living process that AI assistants can connect to, discover available tools, and execute calculations through a secure protocol boundary.

Capabilities

Declare what your server can do:

  • :tools - Execute functions with structured inputs and outputs
  • :resources - Provide data that models can read (files, APIs, databases)
  • :prompts - Offer reusable templates for common interactions
  • :logging - Allow clients to configure verbosity levels

Configure capabilities with options:

use Anubis.Server,
  capabilities: [
    :tools,
    {:resources, subscribe?: true},      # Enable resource update subscriptions
    {:prompts, list_changed?: true}      # Notify when prompts change
  ]

Components

Register tools, resources, and prompts as components:

component MyServer.FileReader           # Auto-named as "file_reader"
component MyServer.ApiClient, name: "api"   # Custom name

Components are modules that implement specific behaviors and are automatically discovered by clients through the protocol.

Server Lifecycle

Your server follows a predictable lifecycle with callbacks you can hook into:

  1. init/2 - Set up initial state when the server starts
  2. handle_request/2 - Process MCP protocol requests from clients
  3. handle_notification/2 - React to one-way client messages
  4. handle_info/2 - Bridge external events into MCP notifications

Most protocol handling is automatic - you typically only implement init/2 for setup and occasionally override other callbacks for custom behavior.

Summary

Callbacks

Handles synchronous calls to the server process.

Handles asynchronous casts to the server process.

Handles completion requests from the client.

Handles non-MCP messages sent to the server process.

Handles incoming MCP notifications from clients.

Handles a prompt get request.

Low-level handler for any MCP request.

Handles a resource read request.

Handles the response from a roots/list request sent to the client.

Handles the response from a sampling/createMessage request sent to the client.

Handles a tool call request.

Called after a client requests a initialize request.

Declares the server's capabilities during initialization.

Provides the server's identity information during initialization.

Specifies which MCP protocol versions this server can speak.

Cleans up when the server process terminates.

Functions

Registers a component (tool, prompt, or resource) with the server.

Checks if the MCP session has been initialized.

Sends a log message to the client.

Sends a progress notification for an ongoing operation.

Sends a prompts list changed notification to connected clients.

Sends a resource updated notification for a specific resource.

Sends a resources list changed notification to connected clients.

Sends a roots/list request to the client.

Sends a sampling/createMessage request to the client.

Sends a tools list changed notification to connected clients.

Types

mcp_error()

@type mcp_error() :: Anubis.MCP.Error.t()

notification()

@type notification() :: map()

progress_step()

@type progress_step() :: number()

progress_token()

@type progress_token() :: String.t() | non_neg_integer()

progress_total()

@type progress_total() :: number()

request()

@type request() :: map()

response()

@type response() :: map()

server_capabilities()

@type server_capabilities() :: map()

server_info()

@type server_info() :: map()

Callbacks

handle_call(request, from, t)

(optional)
@callback handle_call(
  request :: term(),
  from :: GenServer.from(),
  Anubis.Server.Frame.t()
) ::
  {:reply, reply :: term(), Anubis.Server.Frame.t()}
  | {:reply, reply :: term(), Anubis.Server.Frame.t(),
     timeout() | :hibernate | {:continue, arg :: term()}}
  | {:noreply, Anubis.Server.Frame.t()}
  | {:noreply, Anubis.Server.Frame.t(),
     timeout() | :hibernate | {:continue, arg :: term()}}
  | {:stop, reason :: term(), reply :: term(), Anubis.Server.Frame.t()}
  | {:stop, reason :: term(), Anubis.Server.Frame.t()}

Handles synchronous calls to the server process.

This optional callback allows you to handle custom synchronous calls made to your MCP server process using GenServer.call/2. This is useful for implementing administrative functions, status queries, or any synchronous operations that need to interact with the server's internal state.

The callback follows standard GenServer semantics and should return appropriate reply tuples. If not implemented, the Base module provides a default implementation that handles standard MCP operations.

handle_cast(request, t)

(optional)
@callback handle_cast(request :: term(), Anubis.Server.Frame.t()) ::
  {:noreply, Anubis.Server.Frame.t()}
  | {:noreply, Anubis.Server.Frame.t(),
     timeout() | :hibernate | {:continue, arg :: term()}}
  | {:stop, reason :: term(), Anubis.Server.Frame.t()}

Handles asynchronous casts to the server process.

This optional callback allows you to handle custom asynchronous messages sent to your MCP server process using GenServer.cast/2. This is useful for fire-and-forget operations, background tasks, or any asynchronous operations that don't require an immediate response.

The callback follows standard GenServer semantics. If not implemented, the Base module provides a default implementation that handles standard MCP operations.

handle_completion(ref, argument, t)

(optional)
@callback handle_completion(ref :: String.t(), argument :: map(), Anubis.Server.Frame.t()) ::
  {:reply, Anubis.Server.Response.t() | map(), Anubis.Server.Frame.t()}
  | {:error, mcp_error(), Anubis.Server.Frame.t()}

Handles completion requests from the client.

This callback is invoked when a client requests completions for a reference. The reference indicates what type of completion is being requested.

Note: This callback will only be invoked if user declared the completion capability on server definition

handle_info(event, t)

(optional)
@callback handle_info(event :: term(), Anubis.Server.Frame.t()) ::
  {:noreply, Anubis.Server.Frame.t()}
  | {:noreply, Anubis.Server.Frame.t(),
     timeout() | :hibernate | {:continue, arg :: term()}}
  | {:stop, reason :: term(), Anubis.Server.Frame.t()}

Handles non-MCP messages sent to the server process.

While handle_request and handle_notification deal with MCP protocol messages, this callback handles everything else - timer events, messages from other processes, system signals, and any custom inter-process communication your server needs.

This is particularly useful for servers that need to react to external events (like file system changes or database updates) and notify connected clients through MCP notifications. Think of it as the bridge between your Elixir application's internal events and the MCP protocol's notification system.

handle_notification(notification, state)

(optional)
@callback handle_notification(
  notification :: notification(),
  state :: Anubis.Server.Frame.t()
) ::
  {:noreply, new_state :: Anubis.Server.Frame.t()}
  | {:error, error :: mcp_error(), new_state :: Anubis.Server.Frame.t()}

Handles incoming MCP notifications from clients.

Notifications are one-way messages in the MCP protocol - the client informs the server about events or state changes without expecting a response. This fire-and-forget pattern is perfect for status updates, progress tracking, and lifecycle events.

Standard MCP Notifications from Clients:

  • notifications/initialized - Client signals it's ready after successful initialization
  • notifications/cancelled - Client requests cancellation of an in-progress operation
  • notifications/progress - Client reports progress on a long-running operation
  • notifications/roots/list_changed - Client's available filesystem roots have changed

Unlike requests, notifications never receive responses. Any errors during processing are typically logged but not communicated back to the client. This makes notifications ideal for optional features like progress tracking where delivery isn't guaranteed.

The server processes these notifications to update its internal state, trigger side effects, or coordinate with other parts of the system. When using use Anubis.Server, basic notification handling is provided, but you'll often want to override this callback to handle progress updates or cancellations specific to your server's operations.

handle_prompt_get(name, arguments, t)

(optional)
@callback handle_prompt_get(
  name :: String.t(),
  arguments :: map(),
  Anubis.Server.Frame.t()
) ::
  {:reply, messages :: list(), Anubis.Server.Frame.t()}
  | {:error, mcp_error(), Anubis.Server.Frame.t()}

Handles a prompt get request.

This callback is invoked when a client requests a specific prompt template. It receives the prompt name, any arguments to fill into the template, and the current frame.

This callback handles both module-based components (registered with component) and runtime components (registered with Frame.register_prompt/3). For module-based prompts, the framework automatically generates pattern-matched clauses during compilation.

handle_request(request, state)

(optional)
@callback handle_request(request :: request(), state :: Anubis.Server.Frame.t()) ::
  {:reply, response :: response(), new_state :: Anubis.Server.Frame.t()}
  | {:noreply, new_state :: Anubis.Server.Frame.t()}
  | {:error, error :: mcp_error(), new_state :: Anubis.Server.Frame.t()}

Low-level handler for any MCP request.

This is an advanced callback that gives you complete control over request handling. When implemented, it bypasses the automatic routing to handle_tool_call/3, handle_resource_read/2, and handle_prompt_get/3 and all other requests that are handled internally, like tools/list and logging/setLevel.

Use this when you need to:

  • Implement custom request methods beyond the standard MCP protocol
  • Add middleware-like processing before requests reach specific handlers
  • Override the framework's default request routing behavior

Note: If you implement this callback, you become responsible for handling ALL MCP requests, including standard protocol methods like tools/list, resources/list, etc. Consider using the specific callbacks instead unless you need this level of control.

handle_resource_read(uri, t)

(optional)
@callback handle_resource_read(uri :: String.t(), Anubis.Server.Frame.t()) ::
  {:reply, content :: map(), Anubis.Server.Frame.t()}
  | {:error, mcp_error(), Anubis.Server.Frame.t()}

Handles a resource read request.

This callback is invoked when a client requests to read a specific resource. It receives the resource URI and the current frame. Developer's implementation should retrieve and return the resource content.

This callback handles both module-based components (registered with component) and runtime components (registered with Frame.register_resource/3). For module-based resources, the framework automatically generates pattern-matched clauses during compilation.

handle_roots(roots, request_id, t)

(optional)
@callback handle_roots(
  roots :: [map()],
  request_id :: String.t(),
  Anubis.Server.Frame.t()
) ::
  {:noreply, Anubis.Server.Frame.t()}
  | {:stop, reason :: term(), Anubis.Server.Frame.t()}

Handles the response from a roots/list request sent to the client.

This callback is invoked when the client responds to a roots list request initiated by the server. The response contains the available root URIs.

handle_sampling(response, request_id, t)

(optional)
@callback handle_sampling(
  response :: map(),
  request_id :: String.t(),
  Anubis.Server.Frame.t()
) ::
  {:noreply, Anubis.Server.Frame.t()}
  | {:stop, reason :: term(), Anubis.Server.Frame.t()}

Handles the response from a sampling/createMessage request sent to the client.

This callback is invoked when the client responds to a sampling request initiated by the server. The response contains the generated message from the client's LLM.

Parameters

  • response - The response from the client containing:
    • "role" - The role of the generated message (typically "assistant")
    • "content" - The content object with type and data
    • "model" - The model used for generation
    • "stopReason" - Why generation stopped (e.g., "endTurn")
  • request_id - The ID of the original request for correlation
  • frame - The current server frame

Returns

  • {:noreply, frame} - Continue processing
  • {:stop, reason, frame} - Stop the server

Examples

def handle_sampling(response, request_id, frame) do
  %{"content" => %{"text" => text}} = response
  # Process the generated text...
  {:noreply, frame}
end

handle_tool_call(name, arguments, t)

(optional)
@callback handle_tool_call(
  name :: String.t(),
  arguments :: map(),
  Anubis.Server.Frame.t()
) ::
  {:reply, result :: term(), Anubis.Server.Frame.t()}
  | {:error, mcp_error(), Anubis.Server.Frame.t()}

Handles a tool call request.

This callback is invoked when a client calls a specific tool. It receives the tool name, the arguments provided by the client, and the current frame. Developers's implementation should execute the tool's logic and return the result.

This callback handles both module-based components (registered with component) and runtime components (registered with Frame.register_tool/3). For module-based tools, the framework automatically generates pattern-matched clauses during compilation.

init(client_info, t)

(optional)
@callback init(client_info :: map(), Anubis.Server.Frame.t()) ::
  {:ok, Anubis.Server.Frame.t()}

Called after a client requests a initialize request.

This callback is invoked while the MCP handshake starts and so the client may not sent the notifications/initialized message yet. For checking if the notification was already sent and the MCP handshare was successfully completed, you can call the initialized?/1 function.

It receives the client's information and the current frame, allowing you to perform client-specific setup, validate capabilities, or prepare resources based on the connected client.

The client_info parameter contains details about the connected client including its name, version, and any additional metadata. Use this to tailor your server's behavior to specific client implementations or versions.

server_capabilities()

@callback server_capabilities() :: server_capabilities()

Declares the server's capabilities during initialization.

This callback tells clients what features your server supports - which types of resources it can provide, what tools it can execute, whether it supports logging configuration, etc. The capabilities you declare here directly impact which requests the client will send.

When using use Anubis.Server with the capabilities option, this callback is automatically implemented based on your configuration. The macro analyzes your registered components and builds the appropriate capability map, so you rarely need to implement this manually.

server_info()

@callback server_info() :: server_info()

Provides the server's identity information during initialization.

This callback is called during the MCP handshake to identify your server to connecting clients. The information returned here helps clients understand which server they're talking to and ensures version compatibility.

When using use Anubis.Server, this callback is automatically implemented using the name and version options you provide. You only need to implement this manually if you require dynamic server information based on runtime conditions.

supported_protocol_versions()

@callback supported_protocol_versions() :: [String.t()]

Specifies which MCP protocol versions this server can speak.

Protocol version negotiation ensures client and server can communicate effectively. During initialization, the client and server agree on a mutually supported version. This callback returns the list of versions your server understands, typically in order of preference from newest to oldest.

When using use Anubis.Server, this is automatically implemented with sensible defaults covering current and recent protocol versions. Override only if you need to restrict or extend version support for specific compatibility requirements.

terminate(reason, t)

(optional)
@callback terminate(reason :: term(), Anubis.Server.Frame.t()) :: term()

Cleans up when the server process terminates.

This optional callback is invoked when the server process is about to terminate. It allows you to perform cleanup operations, close connections, save state, or release resources before the process exits.

The callback receives the termination reason and the current frame. Any return value is ignored. If not implemented, the Base module provides a default implementation that logs the termination event.

Functions

component(module, opts \\ [])

(macro)

Registers a component (tool, prompt, or resource) with the server.

Examples

# Register with auto-derived name
component MyServer.Tools.Calculator

# Register with custom name
component MyServer.Tools.FileManager, name: "files"

initialized?(frame)

@spec initialized?(Anubis.Server.Frame.t()) :: boolean()

Checks if the MCP session has been initialized.

Returns true if the client has completed the initialization handshake and sent the notifications/initialized message. This is useful for guarding operations that require an active session.

Examples

def handle_info(:check_status, frame) do
  if Anubis.Server.initialized?(frame) do
    # Perform operations requiring initialized session
    {:noreply, frame}
  else
    # Wait for initialization
    {:noreply, frame}
  end
end

send_log_message(frame, level, message, data \\ nil)

@spec send_log_message(
  Anubis.Server.Frame.t(),
  level :: Logger.level(),
  message :: String.t(),
  metadata :: map() | nil
) :: :ok

Sends a log message to the client.

Use this to send diagnostic or informational messages to the client's logging system.

send_progress(frame, progress_token, progress, opts \\ [])

@spec send_progress(Anubis.Server.Frame.t(), progress_token(), progress_step(), opts) ::
  :ok
when opts: [total: progress_total(), message: String.t()]

Sends a progress notification for an ongoing operation.

Use this to update the client on the progress of long-running operations.

send_prompts_list_changed(frame)

@spec send_prompts_list_changed(Anubis.Server.Frame.t()) :: :ok

Sends a prompts list changed notification to connected clients.

Use this when the available prompts have changed (added, removed, or modified). The client will typically re-fetch the prompt list in response.

send_resource_updated(frame, uri, timestamp \\ nil)

@spec send_resource_updated(
  Anubis.Server.Frame.t(),
  uri :: String.t(),
  timestamp :: DateTime.t() | nil
) :: :ok

Sends a resource updated notification for a specific resource.

Use this when the content of a specific resource has changed. Clients that have subscribed to this resource will be notified.

send_resources_list_changed(frame)

@spec send_resources_list_changed(Anubis.Server.Frame.t()) :: :ok

Sends a resources list changed notification to connected clients.

Use this when the available resources have changed (added, removed, or modified). The client will typically re-fetch the resource list in response.

send_roots_request(frame, opts \\ [])

@spec send_roots_request(Anubis.Server.Frame.t(), [
  {:timeout, non_neg_integer() | nil}
]) :: :ok

Sends a roots/list request to the client.

This function queries the client for available root URIs. The client must have declared the roots capability during initialization.

send_sampling_request(frame, messages, opts \\ [])

@spec send_sampling_request(Anubis.Server.Frame.t(), [map()], configuration) :: :ok
when configuration: [
       model_preferences: map() | nil,
       system_prompt: String.t() | nil,
       max_token: non_neg_integer() | nil,
       timeout: non_neg_integer() | nil
     ]

Sends a sampling/createMessage request to the client.

This function is used when the server needs the client to generate a message using its language model. The client must have declared the sampling capability during initialization.

Note: This is an asynchronous operation. The response will be delivered to your handle_sampling/3 callback.

Check https://modelcontextprotocol.io/specification/2025-06-18/client/sampling for more information

Examples

messages = [
  %{"role" => "user", "content" => %{"type" => "text", "text" => "Hello"}}
]

model_preferences = %{"costPriority" => 1.0, "speedPriority" => 0.1, "hints" => [%{"name" => "claude"}]}

:ok = Anubis.Server.send_sampling_request(frame, messages,
  model_preferences: model_preferences,
  system_prompt: "You are a helpful assistant",
  max_tokens: 100
)

send_tools_list_changed(frame)

@spec send_tools_list_changed(Anubis.Server.Frame.t()) :: :ok

Sends a tools list changed notification to connected clients.

Use this when the available tools have changed (added, removed, or modified). The client will typically re-fetch the tool list in response.