Current section
Files
Jump to
Current section
Files
phantom_mcp
README.md
README.md
# Phantom MCP
[](https://hex.pm/packages/phantom_mcp)
[](https://hexdocs.pm/phantom_mcp)
<!-- MDOC -->
MCP (Model Context Protocol) framework for Elixir Plug.
This library provides a complete implementation of the [MCP server specification](https://modelcontextprotocol.io/specification/20.6.03-26/basic/transports) with Plug.
## Installation
Add Phantom to your dependencies:
```elixir
{:phantom_mcp, "~> 0.6.0"},
```
## Stdio Transport (Local Clients)
For local-only clients like Claude Desktop, you can expose your MCP server
over stdin/stdout without needing an HTTP server. Add `Phantom.Stdio` to
your supervision tree:
```elixir
# lib/my_app/application.ex
children = [
{Phantom.Stdio, router: MyApp.MCP.Router}
]
```
For more information about running your MCP server locally with stdio, see
[Phantom.Stdio].
## Streamable HTTP Transport (Remote Clients)
When using with Plug/Phoenix, configure MIME to accept SSE:
```elixir
# config/config.exs
config :mime, :types, %{
"text/event-stream" => ["sse"]
}
```
For Streamable HTTP access to your MCP server, forward
a path from your Plug or Phoenix Router to your MCP router.
<!-- tabs-open -->
### Phoenix
```elixir
defmodule MyAppWeb.Router do
use MyAppWeb, :router
# ...
pipeline :mcp do
plug :accepts, ["json", "sse"]
plug Plug.Parsers,
parsers: [{:json, length: 1_000_000}],
pass: ["application/json"],
json_decoder: JSON
end
scope "/mcp" do
pipe_through :mcp
forward "/", Phantom.Plug,
# Uncomment for remote access from anywhere:
# origins: :all,
# Uncomment for remote access from a specified list:
# origins: ["https://myapp.example"],
# Uncomment for a server bound to localhost, to reject DNS rebinding:
# hosts: ["localhost", "127.0.0.1", "[::1]"],
validate_origin: Mix.env() == :prod,
router: MyApp.MCPRouter
end
end
```
### Plug.Router
```elixir
defmodule MyAppWeb.Router do
use Plug.Router
plug :match
plug Plug.Parsers,
parsers: [{:json, length: 1_000_000}],
pass: ["application/json"],
json_decoder: JSON
plug :dispatch
forward "/mcp",
to: Phantom.Plug,
init_opts: [
router: MyApp.MCP.Router
]
end
```
Finally, import the formatter settings in your `.formatter.exs`. Below is using Phoenix's
generated example as a starting point.
```elixir
[
import_deps: [:ecto, :ecto_sql, :phoenix, :phantom_mcp],
# ...
]
```
<!-- tabs-close -->
Now the fun begins: it's time to define your MCP router that catalogs all your tools, prompts, and resources. When you're creating your MCP server, make sure
you test it with the [MCP Inspector](https://modelcontextprotocol.io/docs/tools/inspector) or your client of choice.
See `Phantom.Plug` for local testing instructions, or `Phantom.Stdio`
for building an escript for direct stdio clients.
First we're define the MCP router:
```elixir
defmodule MyApp.MCP.Router do
@moduledoc """
Provides tools, prompts, and resources to aide in researching
topics and creating research studies using the platform {MyApp}.
"""
use Phantom.Router,
name: "MyApp",
vsn: "1.0", # or Application.spec(:my_app, :vsn),
instructions: @moduledoc
end
```
I used the `@moduledoc` as the same documentation for the client; feel free to separate
the instructions from internal documentation.
> #### Instructions and descriptions are important! {: .neutral}
>
> These instructions and descriptions for tools, prompts, and resources
> will be used by the LLM to determine when and how to use your tooling.
> Don't be too verbose, but also don't have vague instructions.
You likely need to consider authentication, so look at `m:Phantom#module-authentication-and-authorization`
for how to implement it; tldr, implement the `c:Phantom.Router.connect/2` callback and return `{:ok, session}`
upon success.
Now we'll go through each one and show how to respond synchronously or asynchronously.
## Defining Tools
You can define tools with an optional `input_schema` and an optional `output_schema`.
If no `input_schema` is provided, then the client will not know to send arguments to your handlers.
Use a `do` block with `field` declarations to define the input schema. This
generates JSON Schema for clients **and** validates incoming arguments at
dispatch time. See `Phantom.Tool.JSONSchema` for the full type system and
options reference.
```elixir
defmodule MyApp.MCP.Router do
# ...
# the `@description` attribute will automatically be read,
# or you can provide `:description` directly.
@description """
Create a question for the provided Study.
"""
tool :create_question, MyApp.MCP do
field :study_id, :integer, required: true,
description: "The unique identifier for the Study"
field :label, :string, required: true,
description: "The title of the Question"
field :description, :string, required: true,
description: "About one paragraph of detail that defines the question"
end
end
```
Fields support nested objects, arrays, enums, pattern matching, custom
validators, and more:
```elixir
@description "Search for studies"
tool :search_studies do
field :query, :string, required: true
field :limit, :integer, default: 10, maximum: 100
field :status, :string, in: ~w[draft active archived]
field :tags, {:array, :string}
field :filters, :map do
field :category, :string
field :min_price, :number, minimum: 0
end
end
```
You can also use the map-based `input_schema` option for full control over
the JSON Schema. Clients receive the map as given, so any JSON Schema keyword
works; server-side validation only checks `type`, `required`, `properties`,
`enum`, and `items`:
```elixir
tool :create_question, MyApp.MCP,
input_schema: %{
required: ~w[study_id label description],
properties: %{
study_id: %{type: "integer", description: "The Study ID"},
label: %{type: "string", description: "The title"},
description: %{type: "string", description: "The contents"}
}
}
```
Then implement it:
<!-- tabs-open -->
### Synchronously
```elixir
# If outside of the Router, you'll want to `require Phantom.Tool`.
# If implementing in the router, this will already be required.
require Phantom.Tool, as: Tool
def create_question(%{"study_id" => study_id} = params, session) do
changeset = MyApp.Question.changeset(%Question{}, params)
with {:ok, question} <- MyApp.Repo.insert(changeset),
{:ok, uri, resource_template} <-
MyApp.MCP.Router.resource_for(session, :question, id: question.id) do
{:reply, Tool.resource_link(uri, resource_template), session}
else
_ -> {:reply, Tool.error("Invalid paramaters"), session}
end
end
```
### Asynchronously
```elixir
# If outside of the Router, you'll want to `require Phantom.Tool`.
# If implementing in the router, this will already be required.
require Phantom.Tool, as: Tool
def create_question(%{"study_id" => study_id} = params, session) do
Task.async(fn ->
Process.sleep(1000)
changeset = MyApp.Question.changeset(%Question{}, params)
with {:ok, question} <- MyApp.Repo.insert(changeset),
{:ok, uri, resource_template} <-
MyApp.MCP.Router.resource_for(session, :question, id: question.id) do
Session.respond(session, Tool.resource_link(uri, resource_template))
else
_ -> Session.respond(session, Tool.error("Invalid paramaters")))
end
end)
{:noreply, session}
end
```
<!-- tabs-close -->
## Defining Prompts
```elixir
defmodule MyApp.MCP.Router do
# ...
# Prompts may contain arguments. If there are arguments
# you may want to also provide a completion function to
# help the client fill in the argument.
@description """
Review the provided Study and provide meaningful feedback about the
study and let me know if there are gaps or missing questions. We want
a meaningful study that can provide insight to the research goals stated
in the study.
"""
prompt :suggest_questions,
completion_function: :study_complete,
arguments: [
%{
name: "study_id",
description: "The study to review",
required: true
}
]
end
```
Then implement it
<!-- tabs-open -->
### Synchronously
```elixir
require Phantom.Prompt, as: Prompt
def suggest_questions(%{"study_id" => study_id}, session) do
case MyApp.MCP.Router.read_resource(session, :study, id: study_id) do
{:ok, uri, resource} ->
{:reply,
Prompt.response(
assistant: Prompt.embedded_resource(uri, resource),
user: Prompt.text("Wowzers"),
assistant: Prompt.image(File.read!("foo.png")),
user: Prompt.text("Seriously, wowzers")
), session}
error ->
{:error, Phantom.Request.internal_error(), session}
end
end
```
### Asynchronously
```elixir
require Phantom.Prompt, as: Prompt
def suggest_questions(%{"study_id" => study_id}, session) do
Task.async(fn ->
case MyApp.MCP.Router.read_resource(session, :study, id: study_id) do
{:ok, uri, resource} ->
Session.respond(session, Prompt.response(
assistant: Prompt.embedded_resource(uri, resource),
user: Prompt.text("Wowzers"),
assistant: Prompt.image(File.read!("foo.png")),
user: Prompt.text("Seriously, wowzers")
))
error ->
Session.respond(session, Phantom.Request.internal_error())
end
end)
{:noreply, sessin}
end
```
<!-- tabs-close -->
## Defining Resources
Let's define a resource with a resource template:
```elixir
@description """
Read the cover image of a Study to gain some context of the
audience, research goals, and questions.
"""
resource "myapp:///studies/:study_id/cover", :study_cover,
completion_function: :study_complete,
mime_type: "image/png"
@description """
Read the contents of a study. This includes the questions and general
context, which is helpful for understanding research goals.
"""
resource "https://example.com/studies/:study_id/md", :study,
completion_function: :study_complete,
mime_type: "text/markdown"
```
Then implement them:
<!-- tabs-open -->
### Synchronously
```elixir
require Phantom.Resource, as: Resource
def study(%{"study_id" => id} = params, session) do
study = Repo.get(Study, id)
text = Study.to_markdown(study)
{:reply, Resource.text(text), session}
end
def study_cover(%{"study_id" => id} = params, session) do
study = Repo.get(Study, id)
blob = File.read!(study.cover)
{:reply, Resource.blob(blob), session}
end
## Implement the completion handler:
import Ecto.Query
def study_complete("study_id", value, session) do
study_ids = Repo.all(
from s in Study,
select: s.id,
where: like(type(:id, :string), "#{value}%"),
where: s.account_id == ^session.user.account_id,
order_by: s.id,
limit: 101
)
# You may also return a map with more info:
# `%{values: study_ids, has_more: true, total: 1_000_000}`
# If you return more than 100, then Phantom will set `has_more: true`
# and only return the first 100.
{:reply, study_ids, session}
end
```
### Asynchronously
```elixir
require Phantom.Resource, as: Resource
def study(%{"study_id" => id} = params, session) do
Task.async(fn ->
Process.sleep(1000)
study = Repo.get(Study, id)
text = Study.to_markdown(study)
Session.respond(session, Resource.response(Resource.text(text)))
end)
{:noreply, session}
end
def study_cover(%{"study_id" => id} = params, session) do
Task.async(fn ->
Process.sleep(1000)
study = Repo.get(Study, id)
blob = File.read!(study.cover)
Session.respond(session, Resource.response(Resource.blob(blob)))
end)
{:noreply, session}
end
## Implement the completion handler:
import Ecto.Query
def study_complete("study_id", value, session) do
study_ids = Repo.all(
from s in Study,
select: s.id,
where: like(type(:id, :string), "#{value}%"),
where: s.account_id == ^session.user.account_id,
order_by: s.id,
limit: 101
)
# You may also return a map with more info:
# `%{values: study_ids, has_more: true, total: 1_000_000}`
# If you return more than 100, then Phantom will set `has_more: true`
# and only return the first 100.
{:reply, study_ids, session}
end
```
<!-- tabs-close -->
You'll also want to implement `list_resources/2` in your router which is
to provide a list of all available resources in your system and return
resource links to them.
```elixir
@salt "cursor"
def list_resources(cursor, session) do
# Remember to check for allowed resources according to `session.allowed_resource_templates`
# Below is a toy implementation for illustrative purposes.
cursor =
if cursor do
{:ok, cursor} = Phoenix.Token.verify(MyApp.Endpoint, @salt, cursor)
cursor
else
0
end
{_before_cursor, after_cursor} = Enum.split_while(1..1000, fn i -> i < cursor end)
{page, [next | _drop]} = Enum.split(after_cursor, 100)
next_cursor = Phoenix.Token.sign(MyApp.Endpoint, @salt, next)
resource_links =
Enum.map(page, fn i ->
{:ok, uri, spec} = resource_for(session, :study, id: i)
Resource.resource_link(uri, spec, name: "Study #{i}")
end)
{:reply,
Resource.list(resource_links, next_cursor),
session}
end
```
You can notify clients after subscribed resources change. For bulk writes, collect the changed
URIs in your application and notify Phantom once. This lets Phantom group the resources by
subscribed session and lets your authorization callback use one bulk query per session.
```elixir
# Do a bulk write, then collect the affected URIs from its result.
uris =
Enum.map(updated_records, fn record ->
{:ok, uri} = MyApp.MCP.Router.resource_uri(:my_resource, id: record.id)
uri
end)
Phantom.Tracker.notify_resources_updated(uris)
```
Resource subscriptions are allowed by default. For user-scoped resources, override
`authorize_resource_subscriptions/2`. Phantom resolves each URI before invoking the callback and
uses the same callback both when accepting `resources/subscribe` and immediately before sending
update notifications. This second check ensures permission changes take effect after subscription.
```elixir
def authorize_resource_subscriptions(resources, session) do
user = session.assigns.user
allowed_ids =
resources
|> Enum.map(fn {_uri, %{"id" => id}, _template} -> id end)
|> MyApp.Resources.list_authorized_ids(user)
|> MapSet.new()
Enum.filter(resources, fn {_uri, %{"id" => id}, _template} ->
id in allowed_ids
end)
end
```
The callback receives `{uri, path_params, resource_template}` tuples and the authenticated session.
It may return the allowed tuples or just their URI strings. Returning `nil` or `[]`, raising, or
returning an invalid value rejects the resources. URIs that do not resolve to an available resource
template are rejected before the callback runs.
## Eliciting input
Two helpers, picked by how the dev wants to structure the handler.
### Inline blocking
The tool function "awaits" the response in place and continues inline:
```elixir
def my_tool(params, session) do
case Phantom.Session.elicit(session, @elicit_name, await: true) do
{:ok, %{"action" => "accept", "content" => content}} ->
{:reply, Tool.text("Hello \#{content["name"]}"), session}
{:ok, _rejected} ->
{:reply, Tool.error("Rejected"), session}
:not_supported ->
{:reply, Tool.text("Hello stranger"), session}
end
end
```
Under MCP `2026-07-28` the call is answered with an `input_required` result
while the calling process waits; the client's follow-up call, on any node in
the cluster, resumes it and receives the tool's response. The process lives on
the node that started it, so a follow-up after that node restarts, or after
`:timeout` (default 5 minutes), gets an error. The same holds for
`Session.elicit/2` called from a process the handler started, such as a `Task`.
### Re-entry — `Session.elicit/3` (no `:await`)
Call `Session.elicit/3` without `:await` (and pass `:state`) to get the
re-entry pattern: the handler is invoked again with `session.state`
populated on continuation. Same source under both protocols:
```elixir
use Phantom.Router,
name: "MyApp",
secret_key_base: {Application, :fetch_env!, [:my_app, :mcp_secret_key_base]},
request_state_salt: "myapp request_state v1"
@description "Delete a file after confirming with the user"
tool :delete_file do
field :path, :string, required: true
end
# Resume clause — runs on the second invocation under either protocol
def delete_file(
%{"confirm" => "yes"},
%Phantom.Session{state: %{step: :confirming, path: path}} = session
) do
File.rm!(path)
{:reply, Tool.text("Deleted #{path}"), session}
end
def delete_file(%{"confirm" => _}, session),
do: {:reply, Tool.text("Cancelled"), session}
# First-call clause — ask the user
def delete_file(%{"path" => path}, session) do
{:noreply,
Phantom.Session.elicit(
session,
Phantom.Elicit.form(%{
message: "Really delete #{path}?",
requested_schema: [
%{name: "confirm", type: :enum, enum: ["yes", "no"], required: true}
]
}),
state: %{step: :confirming, path: path}
)}
end
```
Under `2026-07-28` the call returns an `input_required` result with an
encrypted `requestState` blob; any node can serve the follow-up. Under
legacy protocols Phantom performs the SSE `elicitation/create` round-trip
and re-invokes the handler — same handler code, no `if protocol_version`
check.
`:secret_key_base` and `:request_state_salt` are both required for
`2026-07-28` — Phantom encrypts the `requestState` blob with `Plug.Crypto`
(no Phoenix dependency) using the salt to derive a domain-specific key.
Each node serving the same router must share both values; clients can hop
nodes freely.
`:secret_key_base` can be a `{module, function, args}` tuple, called each
time Phantom needs the key, so a release can set it in `config/runtime.exs`
instead of compiling it into the router:
```elixir
# config/runtime.exs
config :my_app, mcp_secret_key_base: System.fetch_env!("SECRET_KEY_BASE")
```
With Phoenix, `{MyAppWeb.Endpoint, :config, [:secret_key_base]}` reuses the
endpoint's key.
### Sending the user to a URL
URL elicitation sends the user to a page you host, such as a sign-in or
payment flow, instead of a form in the client. The client must declare
`elicitation: %{"url" => %{}}`; otherwise the call returns `:not_supported`.
The client's `accept` only means the user agreed to open the URL, not that
they finished. Keep your own record of the flow, tied to the signed-in user,
and check it. The simplest shape returns `{:elicitation_required, ...}` until
the record says the user is done; the client opens the URL and calls the tool
again:
```elixir
def connect_account(_params, session) do
user = session.assigns.user
if MyApp.Accounts.connected?(user) do
{:reply, Tool.text("Connected"), session}
else
elicitation_id = UUIDv7.generate()
MyApp.Accounts.start_connect(user, elicitation_id)
{:elicitation_required,
[
Phantom.Elicit.url(%{
message: "Connect your account",
url: "https://myapp.example/connect/#{elicitation_id}",
elicitation_id: elicitation_id
})
]}
end
end
```
The page must check that the signed-in user is the one who started the flow.
Otherwise someone could send their link to another user and have that user
connect an account for them:
```elixir
def connect(conn, %{"elicitation_id" => elicitation_id}) do
case MyApp.Accounts.get_connect(elicitation_id) do
%{user_id: user_id} = flow when user_id == conn.assigns.current_user.id ->
MyApp.Accounts.complete_connect(flow)
render(conn, :connected)
_ ->
conn |> put_status(:forbidden) |> render(:wrong_user)
end
end
```
See `Phantom.Elicit` for what each protocol version sends, and for waiting
inline with `Phantom.Session.elicit/3`.
## What PhantomMCP supports
Phantom will implement these MCP requests on your behalf:
- `initialize`. Phantom will detect what capabilities are available to the client based on the provided tooling defined in the Phantom router.
- `prompts/list` list either the allowed prompts provided in the `connect/2` callback, or all prompts by default. To disable, return `allow_prompts(session, [])` in the `connect/2` callback.
- `prompts/get` dispatch the request to your handler if allowed. Read more in `Phantom.Prompt`.
- `resources/list` dispatch to your MCP router. By default it will be an empty list until you implement it. Read more in `Phantom.Resource`.
- `resource/templates/list` list either the allowed resources as provided in the `connect/2` callback or all resource templates by default. To disable, return `allow_resource_templates(session, [])` in the `connect/2` callback. Read more in `Phantom.ResourceTemplate`.
- `resources/read` dispatch the request to your handler. `Phantom.Resource`.
- `resources/subscribe` available if the MCP router is configured with `pubsub`. To notify of updates, prefer `Phantom.Tracker.notify_resources_updated(uris)`; `notify_resource_updated(uri)` remains available for individual changes.
- `resources/unsubscribe` see above.
- `logging/setLevel` available if the MCP router is configured with `pubsub`. Logs can be sent to client with `Session.log_{level}(session, map_content)`. [See docs](https://modelcontextprotocol.io/specification/20.6.03-26/server/utilities/logging#log-levels).
- `tools/list` list either the allowed tools as provided in the `connect/2` callback or all tools by default. To disable, return `allow_tools(session, [])` in the `connect/2` callback.
- `tools/call` dispatch the request to your handler. Read more in `Phantom.Tool`.
- `completion/complete` dispatch the request to your completion handler for the given prompt or resource.
- `notification/*` no-op.
- `ping` pong
- `notifications/resources/list_changed` - The server informs the client the list of resources has updated. This is not done automatically; you will need to trigger this with `Phantom.Tracker.notify_resource_list/0`, but also be mindful of what resources the session may have access to.
- `notifications/prompts/list_changed` - The server informs the client the list of prompts has updated. This is triggered when `Phantom.Cache.add_prompt/2` is called.
- `notifications/tools/list_changed` - The server informs the client the list of tools has updated. This is triggered when `Phantom.Cache.add_tool/2` is called.
- `elicitation/create` - The server requests input from
the client in order to complete a request the client has made of it.
Under MCP `2026-07-28` the server-initiated SSE push is replaced by an
`input_required` result + encrypted `requestState`; both flows go through
`Phantom.Session.elicit/3` and behave identically from the tool's point
of view.
Phantom **does not yet support these methods**:
- `roots/list` - The server requests the client to provide a list of files available for interaction. This is like `resources/list` but for the client.
- `sampling/createMessage` - The server requests the client to query their LLM and provide its response. This is for human-in-the-loop agentic actions and could be leveraged when the client requests a prompt from the server.
## Batched Requests
Batched requests will also be handled transparently. **please note** there is not an abstraction for efficiently providing these as a group to your handler. Since the MCP specification is deprecating batched request support in the next version, there is no plan to make this more efficient.
## Authentication and Authorization
Phantom does not implement authentication on its own. MCP applications needing authentication should investigate OAuth provider solutions like [Oidcc](https://hex.pm/packages/oidcc) or [Boruta](https://hex.pm/packages/boruta) or [ExOauth2Provider](https://hex.pm/packages/ex_oauth2_provider) and configure the route to serve a discovery endpoint.
1. [MCP authentication and discovery](https://modelcontextprotocol.io/specification/20.6.06-18/basic/authorization) is not handled by Phantom itself. You will need to implement OAuth2 and provide the discovery mechanisms as described in the specification. In the `connect/2` callback you can return `{:unauthorized, www_authenticate_info}` or `{:forbidden, "error message"}` to inform the client of how to move forward. An `{:ok, session}` result will imply successful auth.
2. Once the authentication flow has been completed, a request to the MCP router should land with an authorization header that can be received and verified in the `connect/2` callback of your MCP router.
3. You may also decide to limit the available tools, prompts, or resources depending on your authorization rules. An example is below.
```elixir
defmodule MyApp.MCP.Router do
use Phantom.Router,
name: "MyApp",
vsn: "1.0"
require Logger
def connect(session, %{headers: auth_info}) do
# The `auth_info` will depend on the adapter, in this case it's from
# Plug, so it will contain query parameters and request headers.
with {:ok, user} <- MyApp.authenticate(conn, auth_info),
{:ok, my_session_state} <- MyApp.load_session(session.id) do
{:ok,
session
|> assign(some_state: my_session_state, user: user)
|> limit_for_plan(user.plan)}
else
:not_found ->
# See `Phantom.Plug.www_authenticate/1`
{:unauthorized, %{
method: "Bearer",
resource_metadata: "https://myapp.com/.well-known/oauth-protected-resource"
}}
:not_allowed ->
{:forbidden, "Please upgrade plan to use MCP server"}
end
end
defp limit_for_plan(session, :ultra), do: session
defp limit_for_plan(session, :basic) do
# allow-list tools by stringified name. The name is either supplied as the `name` when defining it, or the stringified function name.
session
|> Phantom.Session.allowed_tools(~w[create_question])
|> Phantom.Session.allowed_resource_templates(~w[study])
end
```
## Optional callbacks
There are several optional callbacks to help you hook into the lifecycle of the connections.
- `c:Phantom.Router.disconnect/1` means the request has closed, not that the session is finished.
- `c:Phantom.Router.terminate/1` means the session has finished and the client doesn't intend to resume it. Phantom closes the session's streams but doesn't remember the session; to answer later requests with HTTP 404 as the spec requires, record it here and return `{:not_found, message}` from `c:Phantom.Router.connect/2`.
For Telemetry, please see `m:Phantom.Plug#module-telemetry` and `m:Phantom.Router#module-telemetry` for emitted telemetry hooks.
## Distributed tracing
Under MCP `2026-07-28`, clients carry W3C Trace Context — `traceparent`,
`tracestate`, and `baggage` — in the request's `_meta`. Phantom automatically
extracts these and surfaces them on the `[:phantom, :dispatch]` telemetry
span under `metadata.trace_context`. Wire your tracer to the event.
Example with OpenTelemetry:
```elixir
# In your application's start callback:
:telemetry.attach(
"phantom-otel-dispatch",
[:phantom, :dispatch, :start],
&MyApp.Telemetry.handle_dispatch/4,
nil
)
defmodule MyApp.Telemetry do
def handle_dispatch(_event, _measurements, %{method: method, trace_context: ctx}, _config) do
# Extract the upstream W3C trace context into OpenTelemetry's process state
:otel_propagator_text_map.extract(Enum.map(ctx, fn {k, v} -> {Atom.to_string(k), v} end))
# Start a child span for this dispatch
OpenTelemetry.Tracer.start_span("mcp:#{method}")
end
end
```
For other tracers, the pattern is the same: attach to `[:phantom, :dispatch, :start]`
and read `metadata.trace_context` — it's a map with `:traceparent`, `:tracestate`,
and `:baggage` keys (only the ones present in the request are included).
## Use with Load Balancers and Reverse Proxies
Phantom is designed to run behind a load balancer. The shape of that
deployment depends on which MCP protocol versions you serve:
| Client protocol | Sticky session required? | What the LB needs to do |
|---|---|---|
| `≤ 2025-11-25` (legacy) | Yes, *or* run `Phantom.Tracker` for cross-node routing | Hash on `mcp-session-id` header, OR enable cross-node session replication via `Phantom.PubSub` |
| `2026-07-28` (stateless core) | No | Plain round-robin |
You can serve both from the same router — the LB just needs to handle the
legacy case correctly.
### Sticky sessions for legacy clients
Route by the `mcp-session-id` request header. The header is present on
every legacy POST after `initialize`, and on the persistent SSE GET stream.
Nginx:
```nginx
upstream phantom_mcp {
hash $http_mcp_session_id consistent;
server app1.internal:4000;
server app2.internal:4000;
}
```
HAProxy:
```
backend phantom_mcp
balance hdr(mcp-session-id)
server app1 app1.internal:4000 check
server app2 app2.internal:4000 check
```
**If you can't configure header-based hashing** (cloud LBs without L7
hashing, for example), run `Phantom.Tracker` + `Phoenix.PubSub` in your
release. Any node can then route session-bound traffic to the owning node
internally. See the "Persistent Streams" section below.
### Stateless clients need no LB configuration
Under MCP `2026-07-28`, every request is self-contained. The encrypted
`requestState` blob carries continuation state across requests; any node
that shares `:secret_key_base` and `:request_state_salt` can serve any request. Use round-robin or
least-connections; sticky routing is wasted complexity.
### L7 routing with MCP `2026-07-28` headers
SEP-2243 requires clients on `2026-07-28` to mirror routing fields from
the JSON-RPC body into HTTP headers so load balancers, proxies, and WAFs
can route on the operation without parsing the body:
- `Mcp-Method` — mirrors `method`. Required on every POST.
- `Mcp-Name` — mirrors `params.name` (for `tools/call`, `prompts/get`)
or `params.uri` (for `resources/read`).
Header names are case-insensitive; values are case-sensitive (so
`Mcp-Method: TOOLS/CALL` does *not* match a body method of `tools/call`).
Phantom validates these server-side and rejects requests where the
headers and body disagree (or where required headers are missing) with
HTTP `400 Bad Request` and JSON-RPC error code `-32001 HeaderMismatch`.
Legacy clients negotiating an older protocol version are exempt.
### Response caching
When a handler annotates its result with `Phantom.Request.with_cache/2`,
the response gains top-level `ttlMs` and `cacheScope` fields modeled on
HTTP `Cache-Control`. A reverse proxy can read these and cache
accordingly. For Nginx with `proxy_cache`, you'd write a small Lua snippet
(or use OpenResty) to read the JSON body's `result.ttlMs` and
`result.cacheScope` and call `ngx.cache_control` accordingly.
For simpler setups: just look at `Cache-Control: ...` headers your proxy
emits. Phantom doesn't currently translate `ttlMs`/`cacheScope` into HTTP
`Cache-Control` response headers automatically — they ride in the JSON
result for MCP-aware clients. If you want HTTP-level caching, configure
your LB to translate, or wrap `Phantom.Plug` in another Plug that mirrors
the values into headers.
| Annotation | Reverse proxy / CDN behavior |
|---|---|
| `cacheScope: "public", ttlMs: 60_000` | Cache and serve to any user for 60s |
| `cacheScope: "private", ttlMs: 60_000` | Client-local cache only; shared caches must pass through |
Watch out: marking a user-specific response `"public"` lets one user's
result be served to another from a shared cache. Default to omitting the
hints unless you're sure the result is repeatable for the user. See
"Defining Resources" / `Phantom.Request.with_cache/2` for the API.
### Origin validation behind a proxy
By default `Phantom.Plug` validates the `Origin` header against the
configured `:origins` allowlist. Behind a TLS-terminating proxy, the
forwarded request still carries the original `Origin` — no extra config
needed. If your proxy stripping `Origin` is a concern, configure
`Plug.RewriteOn` or equivalent to restore it from `X-Forwarded-Host`
before Phantom runs.
### Health checks
`Phantom.Plug` doesn't expose a dedicated health endpoint. Either:
- Have your LB GET a separate Plug path (e.g. `/health`) defined in your
app's main router, *not* forwarded to `Phantom.Plug`.
- Or send a JSON-RPC `ping` request to `/mcp` and check for `{"jsonrpc":
"2.0", "result": "pong"}` — Phantom handles `ping` natively.
## Persistent Streams (legacy protocols, ≤ 2025-11-25)
> Under MCP `2026-07-28` (stateless core), persistent SSE streams and
> `mcp-session-id` are gone — every request is self-contained and any node
> can serve any call. The setup below applies to clients on earlier protocol
> versions; you can run both side-by-side from the same router.
MCP supports SSE streams to get notifications allow resource subscriptions.
To support this, Phantom needs to track connection pids in your cluster and uses
Phoenix.Tracker (`phoenix_pubsub`) to do this.
MCP defines a "Streamable HTTP" protocol, which is typical HTTP and SSE connections but with a certain behavior. MCP will typically have multiple connections to facilitate requests:
1. `POST` for every command, such as `tools/call` or `resources/read`. Phantom will open an SSE stream and then immediately close the connection once work has completed.
2. `GET` to start an SSE stream for any events such as logs and notifications. There is no work to complete with these requests, and therefore is just a "channel" for receiving server requests and notifications. The connection will remain open until either the client or server closes it.
All connections may provide an `mcp-session-id` header to resume a session.
**Not yet supported** is the ability to resume broken connections with missed messages with the `last-event-id` header, however this is planned to be supported with a ttl-expiring distributed circular buffer.
To make Phantom distributed, start the `Phantom.Tracker` and pass in your pubsub module to the `Phantom.Plug` options:
```elixir
# Add to your application supervision tree:
{Phoenix.PubSub, name: MyApp.PubSub},
{Phantom.Tracker, [name: Phantom.Tracker, pubsub_server: MyApp.PubSub]},
```
Adjust the Phoenix router or Plug.Router options to include the PubSub server
<!-- tabs-open -->
### Phoenix
```elixir
forward "/", Phantom.Plug,
router: MyApp.MCP.Router,
pubsub: MyApp.PubSub
```
### Plug.Router
```elixir
forward "/mcp", to: Phantom.Plug, init_opts: [
router: MyApp.MCP.Router,
pubsub: MyApp.PubSub
]
```
<!-- tabs-close -->