Anubis.Server behaviour (anubis_mcp v1.6.1)

Copy Markdown

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
  @moduledoc "Add two numbers"

  use Anubis.Server.Component, type: :tool

  schema do
    field :a, :number, required: true
    field :b, :number, required: true
  end

  def execute(%{a: a, b: b}, _frame) do
    {:ok, a + b}
  end
end

# In your supervision tree
children = [{MyServer, transport: :stdio}]
Supervisor.start_link(children, strategy: :one_for_one)

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.

Sending Notifications

Notification functions use send(self(), ...) and must be called from within the Session process (i.e., inside callbacks). For sending from external processes or tasks, use send/2 with the session PID directly.

# Inside a callback:
def handle_info(:data_changed, frame) do
  Anubis.Server.send_tools_list_changed()
  {:noreply, frame}
end

Summary

Callbacks

Handles incoming MCP notifications from clients.

Handles a prompt get request.

Low-level handler for any MCP request.

Handles a resource read request.

Called when a session is being auto-recovered after expiry.

Handles a tool call request.

Called after a client requests a initialize request.

Returns optional instructions describing how to use the server and its features.

Functions

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

Sends an elicitation/create request to the client.

Sends a log message to the client.

Sends a progress notification for an ongoing operation.

Sends a prompts list changed notification.

Sends a resource updated notification for a specific resource.

Sends a resources list changed notification.

Sends a roots/list request to the client.

Sends a sampling/createMessage request to the client.

Sends a notifications/tasks/status notification with the current state of the given task.

Sends a tools list changed notification.

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()}

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()}

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()}

handle_elicitation(response, request_id, t)

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

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()}

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.

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.

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.

When implemented, it bypasses automatic routing to specific handlers.

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.

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()}

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()}

handle_session_expired(session_id, t)

(optional)
@callback handle_session_expired(session_id :: String.t(), Anubis.Server.Frame.t()) ::
  {:ok, Anubis.Server.Frame.t()}
  | {:ok, client_info :: map(), Anubis.Server.Frame.t()}
  | {:error, reason :: term()}

Called when a session is being auto-recovered after expiry.

Invoked during auto_initialize/1 instead of the normal client handshake. Receives the session ID and the current frame (pre-populated from the session store if one is configured).

Return values:

  • {:ok, frame} — accept recovery using synthetic client info
  • {:ok, client_info, frame} — accept recovery and supply real client info
  • {:error, reason} — reject recovery; the client receives an internal error

If this callback is not implemented, the default behavior is unchanged: synthetic client info is used and init/2 is called normally.

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.

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 handshake was successfully completed, you can check the context.initialized field in the frame.

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.

server_capabilities()

@callback server_capabilities() :: server_capabilities()

server_info()

@callback server_info() :: server_info()

server_instructions()

(optional)
@callback server_instructions() :: String.t() | nil

Returns optional instructions describing how to use the server and its features.

This can be used by clients to improve the LLM's understanding of available tools, resources, etc. It can be thought of like a "hint" to the model. For example, this information MAY be added to the system prompt.

Return nil to omit the instructions field from the initialize response.

supported_protocol_versions()

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

terminate(reason, t)

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

Functions

component(module, opts \\ [])

(macro)

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

send_elicitation_request(message, requested_schema, opts \\ [])

@spec send_elicitation_request(String.t(), map(), configuration) ::
  :ok | {:error, term()}
when configuration: [{:timeout, non_neg_integer() | nil}]

Sends an elicitation/create request to the client.

Per the MCP 2025-06-18 specification, the server provides a human-readable message and a restricted-subset JSON requested_schema describing the expected user input. The client presents this to the user and returns one of three actions: accept (with content matching the schema), decline, or cancel.

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

The requested_schema is validated synchronously before any wire I/O. The client must advertise the elicitation capability or the call returns {:error, :capability_not_supported} after enqueueing.

Schema Subset

  • Top level must be %{"type" => "object", "properties" => %{...}}
  • Properties may declare "type" of "string", "number", "integer", "boolean", or use "enum" (string-only)
  • String properties may set "format" of "email", "uri", "date", or "date-time"

Per the spec, servers MUST NOT request sensitive information through elicitation.

Example

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

  @impl true
  def execute(_args, frame) do
    Anubis.Server.send_elicitation_request("What's your name?", %{
      "type" => "object",
      "properties" => %{
        "name" => %{"type" => "string", "minLength" => 1}
      },
      "required" => ["name"]
    })

    {:reply, "asked for name", frame}
  end
end

send_log_message(level, message, data \\ nil)

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

Sends a log message to the client.

Must be called from within a Session callback — see send_resources_list_changed/0 for details.

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

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

Sends a progress notification for an ongoing operation.

send_prompts_list_changed()

@spec send_prompts_list_changed() :: :ok

Sends a prompts list changed notification.

Must be called from within a Session callback — see send_resources_list_changed/0 for details.

send_resource_updated(uri, timestamp \\ nil)

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

Sends a resource updated notification for a specific resource.

Subscription-gated: only emits if the current session has previously received a resources/subscribe request for this URI. Calls for unsubscribed URIs are silently dropped.

Must be called from within a Session callback — see send_resources_list_changed/0 for details.

send_resources_list_changed()

@spec send_resources_list_changed() :: :ok

Sends a resources list changed notification.

Must be called from within a Session callback — the current process must be the Session GenServer. Calling from outside a callback will silently lose the message.

For external processes, use send(session_pid, {:send_notification, "notifications/resources/list_changed", %{}}).

send_roots_request(opts \\ [])

@spec send_roots_request([{:timeout, non_neg_integer() | nil}]) :: :ok

Sends a roots/list request to the client.

send_sampling_request(messages, opts \\ [])

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

Sends a sampling/createMessage request to the client.

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

send_task_status(task_id)

@spec send_task_status(task_id :: String.t()) :: :ok

Sends a notifications/tasks/status notification with the current state of the given task.

Must be called from within a Session callback — see send_resources_list_changed/0 for details.

Per spec (2025-11-25), receivers MAY send these notifications when a task's status changes; they are optional and requestors MUST NOT rely on them. This helper looks up the task in the configured task store and emits the full Task projection.

send_tools_list_changed()

@spec send_tools_list_changed() :: :ok

Sends a tools list changed notification.

Must be called from within a Session callback — see send_resources_list_changed/0 for details.