Building a Server

An MCP server exposes parts of your application to AI clients. In Anubis a server is one module that declares identity and capabilities, plus one module per tool, resource, or prompt. This guide builds up each piece.

The server module

defmodule MyApp.Server do
  use Anubis.Server,
    name: "my-app",
    version: "1.0.0",
    capabilities: [:tools, :resources, :prompts]

  component MyApp.ProductSearch
  component MyApp.ConfigResource
  component MyApp.BugReportPrompt
end

The name and version identify your server to clients during the handshake. The capabilities list declares what the server offers. Available capabilities are :tools, :resources, :prompts, :logging, and :completion. Some accept options:

use Anubis.Server,
  name: "my-app",
  version: "1.0.0",
  capabilities: [
    :tools,
    {:resources, subscribe?: true},
    {:prompts, list_changed?: true}
  ]

Each component line registers a module. The component name defaults to the module name converted to snake case, so MyApp.ProductSearch becomes product_search. Pass name: to override it:

component MyApp.ProductSearch, name: "search"

Start the server under your supervision tree with a transport:

children = [
  {MyApp.Server, transport: :stdio}
]

The Transports guide covers STDIO, Streamable HTTP, and Phoenix integration. The rest of this guide is transport independent.

Tools

A tool is a module that declares an input schema and implements execute/2:

defmodule MyApp.ProductSearch do
  @moduledoc "Search the product catalog"

  use Anubis.Server.Component, type: :tool

  alias Anubis.Server.Response

  schema do
    field :query, :string, required: true
    field :limit, :integer, default: 10
    field :category, :string
  end

  @impl true
  def execute(%{query: query} = params, frame) do
    products =
      query
      |> MyApp.Catalog.search()
      |> maybe_filter(params[:category])
      |> Enum.take(params.limit)

    {:reply, Response.json(Response.tool(), products), frame}
  end

  defp maybe_filter(products, nil), do: products
  defp maybe_filter(products, category), do: Enum.filter(products, &(&1.category == category))
end

The schema block does double duty. It generates the JSON Schema that clients see when they list tools, and it validates incoming parameters before execute/2 runs. Your function receives an atom-keyed map that already passed validation.

Fields support types (:string, :integer, :number, :boolean, :enum, {:list, type}), constraints (required, default, min, max, min_length, max_length, regex, values), and metadata (description, format). Nested structures use embeds_one and embeds_many:

schema do
  field :query, :string, required: true, description: "Full text search query"
  field :sort, :enum, values: ["price", "name"], default: "name"

  embeds_one :pagination do
    field :page, :integer, min: 1, default: 1
    field :per_page, :integer, min: 1, max: 100, default: 20
  end
end

Responses

Build tool responses with Anubis.Server.Response. Start from Response.tool() and pipe into content builders:

Response.text(Response.tool(), "plain text")

Response.json(Response.tool(), %{count: 3, items: items})

Response.tool()
|> Response.structured(%{count: 3})

Response.image(Response.tool(), base64_data, "image/png")

Response.error(Response.tool(), "Query failed: timeout")

Response.json/2 encodes the data and adds it as text content. Response.structured/2 sets the structuredContent field defined by the protocol, which pairs with an output_schema block in the component if you want clients to validate your output.

Response.error/2 produces a tool-level error, which tells the model the call failed and why. Reserve it for domain failures the model could react to. For protocol-level failures return an Anubis.MCP.Error instead:

def execute(params, frame) do
  case MyApp.Catalog.search(params.query) do
    {:ok, products} ->
      {:reply, Response.json(Response.tool(), products), frame}

    {:error, :backend_down} ->
      {:error, Anubis.MCP.Error.execution("catalog backend unavailable"), frame}
  end
end

Resources

Resources give clients read access to data. Each resource has a URI and implements read/2:

defmodule MyApp.ConfigResource do
  @moduledoc "Current application configuration"

  use Anubis.Server.Component,
    type: :resource,
    uri: "config://app/settings",
    mime_type: "application/json"

  alias Anubis.Server.Response

  @impl true
  def read(_params, frame) do
    config = %{
      environment: Application.get_env(:my_app, :environment),
      version: to_string(Application.spec(:my_app, :vsn))
    }

    {:reply, Response.json(Response.resource(), config), frame}
  end
end

Clients discover resources through resources/list and fetch them by URI through resources/read. Text content goes through Response.text/2, binary content through Response.blob/2.

Prompts

Prompts are message templates. The client fetches one with arguments, and your module returns the messages to hand to the model:

defmodule MyApp.BugReportPrompt do
  @moduledoc "Structure a bug report from user input"

  use Anubis.Server.Component, type: :prompt

  alias Anubis.Server.Response

  schema do
    field :title, :string, required: true
    field :severity, :enum, values: ["low", "medium", "high"], default: "low"
  end

  @impl true
  def get_messages(%{title: title, severity: severity}, frame) do
    response =
      Response.prompt()
      |> Response.user_message("""
      File a bug report titled "#{title}" with severity #{severity}.
      Ask me for reproduction steps if anything is unclear.
      """)

    {:reply, response, frame}
  end
end

Response.user_message/2, Response.assistant_message/2, and Response.system_message/2 append messages in order.

Descriptions

Clients pick tools by reading their descriptions, so treat them as part of your interface. By default a component's description is its @moduledoc. When the description depends on runtime state, implement description/0 instead:

defmodule MyApp.WeatherTool do
  @moduledoc false

  use Anubis.Server.Component, type: :tool

  @impl true
  def description do
    interval = Application.get_env(:my_app, :weather_cache_minutes, 15)
    "Current weather for a city. Data refreshes every #{interval} minutes."
  end
end

A good description says what the component does and when to use it, in a sentence or two. Field-level description: options serve the same purpose for parameters.

Server state and the frame

Every callback receives a %Anubis.Server.Frame{}. The frame carries two things you will use often:

frame.assigns is your own state for the session, managed like assigns in a Plug.Conn or LiveView socket.
frame.context holds request context: the session id, the client info sent during the handshake, and for HTTP transports the request headers, remote IP, and validated auth claims.

The init/2 callback runs when a client starts the handshake and is the place to set up assigns:

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

  component MyApp.ProductSearch

  @impl true
  def init(_client_info, frame) do
    {:ok, assign(frame, allowed_categories: MyApp.Catalog.categories())}
  end
end

Components then read and update the frame:

def execute(params, frame) do
  categories = frame.assigns.allowed_categories
  frame = assign(frame, :last_query, params.query)

  {:reply, Response.json(Response.tool(), search(params, categories)), frame}
end

Each connected client gets its own session process with its own frame, so assigns are naturally isolated per session.

Asking the client for input

In the stateless era (2026-07-28), a server that needs something from the user or the client's model mid-request does not send its own request: a tools/call, prompts/get or resources/read handler returns an Anubis.Server.InputRequired instead of a response. The client gathers the answers and retries the original request with them, and the handler reads them from the frame.

alias Anubis.Server.{Frame, InputRequired, Response}

@impl true
def execute(%{"to" => to, "text" => text}, frame) do
  case Frame.input_response(frame, "confirm") do
    %{"action" => "accept"} ->
      MyApp.Messages.send(to, text)
      {:reply, Response.text(Response.tool(), "Sent."), frame}

    _not_yet ->
      input =
        InputRequired.new()
        |> InputRequired.elicit("confirm", "Send \"#{text}\" to #{to}?", %{"type" => "object"})

      {:reply, input, frame}
  end
end

InputRequired.elicit/4, sample/3 and list_roots/2 add the three kinds of input request. Each needs the matching client capability, including its mode or subfeature; an unsupported request is refused with -32021 and the required capabilities. Form elicitation requires elicitation.form (the legacy elicitation: {} also permits form); URL-only clients cannot receive a form request. Sampling with tools or toolChoice requires sampling.tools. Frame.client_supports?/2 checks only top-level presence, so handlers offering alternatives also inspect the subfeatures in frame.context.client_capabilities.
A client may answer only part of what was asked, or nothing, and may never retry. Ask again for what is missing rather than failing; answers the handler did not ask for are simply there to ignore.
InputRequired.state/2 attaches a term that comes back on the retry as Frame.request_state/1, for handlers that work in several rounds. It travels signed with HMAC-SHA256 and bound to the authenticated subject, to the tool, prompt or resource, and to a ten-minute expiry (:request_state_ttl); a state that fails any of these is refused with -32602 before the handler runs. Signing needs a secret of at least 32 bytes:
```
config :anubis_mcp, request_state_secret: System.fetch_env!("MCP_REQUEST_STATE_SECRET")
```
A server that authenticates outside Anubis.Server.Authorization has no subject on the frame, so it puts its own principal in the state and checks it on the retry.
Handshake-era requests cannot receive this result; a handler serving both eras checks frame.context.protocol_version before returning one.

Notifications

Servers can push notifications to connected clients. Anubis.Server.Notifier reads the session pid off the frame, so its helpers work from any process — including a task spawned inside a tool callback:

Notifier.tools_list_changed(frame)
Notifier.resources_list_changed(frame)
Notifier.prompts_list_changed(frame)
Notifier.resource_updated(frame, uri)
Notifier.log_message(frame, :info, "reindex finished")
Notifier.progress(frame, token, 50, total: 100)

The list-changed notifications require the matching capability option, for example {:tools, list_changed?: true}. A common pattern bridges application events into MCP notifications through handle_info/2:

@impl true
def init(_client_info, frame) do
  Phoenix.PubSub.subscribe(MyApp.PubSub, "catalog")
  {:ok, frame}
end

@impl true
def handle_info({:catalog_updated, _}, frame) do
  Notifier.resources_list_changed(frame)
  {:noreply, frame}
end

The progress token for Notifier.progress/4 is the one the client sent on the request, which the session exposes as frame.context.request_meta:

@impl true
def execute(_args, frame) do
  token = frame.context.request_meta["progressToken"]
  MyApp.Job.run(fn step -> Notifier.progress(frame, token, step.done, total: step.total) end)
  {:reply, Response.text(Response.tool(), "queued"), frame}
end

Anubis.Server.send_tools_list_changed/0 and its siblings use send(self(), ...), so they only fire when called from the session process itself. Prefer Notifier in callbacks.

Next steps

Transports shows how to serve this over STDIO or HTTP, including from a Phoenix app.
Authorization adds OAuth 2.1 bearer token validation and per-component scopes.
Testing covers testing components without a running transport.

← Previous Page Introduction

Next Page → Building a Client