技术文档摘要

Phantom MCP

<!-- MDOC -->

MCP（模型上下文协议）框架用于Elixir Plug。

此库提供了完整的MCP服务器规范实现，使用Plug。

安装

在依赖项中添加Phantom：

  {:phantom_mcp, "~> 0.3.2"},

当与Plug/Phoenix一起使用时，配置MIME以接受SSE：

# config/config.exs
config :mime, :types, %{
  "text/event-stream" => ["sse"]
}

为了通过流式HTTP访问您的MCP服务器，请从您的Plug或Phoenix路由器转发路径到您的MCP路由器。

<!-- tabs-open -->

Phoenix

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,
      # 取消注释以允许来自任何地方的远程访问：
      # origins: :all,
      # 取消注释以允许来自指定列表的远程访问：
      # origins: ["https://myapp.example"],
      validate_origin: Mix.env() == :prod,
      router: MyApp.MCPRouter
  end
end

Plug.Router

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

最后，在您的.formatter.exs中导入格式化设置。以下示例使用Phoenix生成的示例作为起点。

[
  import_deps: [:ecto, :ecto_sql, :phoenix, :phantom_mcp],
  # ...
]

<!-- tabs-close -->

现在开始有趣的部分：定义您的MCP路由器，该路由器记录了所有工具、提示和资源。在创建您的MCP服务器时，请确保使用MCP检查器或您选择的客户端进行测试。

对于本地测试，您可以使用mcp-remote将仅限本地的客户端代理到您的Phantom驱动的MCP服务器，无论是在本地还是远程托管。不要使用mcp-proxy，因为它设计用于基于旧版SSE的MCP服务器（Phantom使用新的流式HTTP行为）。

首先我们定义MCP路由器：

defmodule MyApp.MCP.Router do
  @moduledoc """
  提供工具、提示和资源，以帮助使用平台{MyApp}研究主题和创建研究项目。
  """

  use Phantom.Router,
    name: "MyApp",
    vsn: "1.0",
    instructions: @moduledoc
end

我使用@moduledoc作为客户端的相同文档；您可以自由地将说明与内部文档分开。

说明和描述很重要！ {: .neutral}

这些工具、提示和资源的说明和描述将被LLM用来确定何时以及如何使用您的工具。不要太冗长，但也不要模棱两可。

您可能需要考虑身份验证，因此请查看m:Phantom#module-authentication-and-authorization以了解如何实现它；简而言之，在c:Phantom.Router.connect/2回调中实现并返回{:ok, session}。

接下来我们将逐一展示如何同步或异步响应。

定义工具

您可以定义具有可选input_schema和可选output_schema的工具。如果没有提供input_schema，则客户端将不知道向处理程序发送参数。

defmodule MyApp.MCP.Router do
  # ...

  # 定义可用工具
  # `@description`属性将自动读取，或者您可以直接提供`:description`。
  @description """
  为提供的研究创建一个问题。
  """
  tool :create_question,
    # 提供一个处理程序。如果未提供，则假定当前模块。
    MyApp.MCP,
    # 提供`input_schema`。
    input_schema: %{
      required: ~w[description label study_id],
      properties: %{
        study_id: %{
          type: "integer",
          description: "研究的唯一标识符"
        },
        label: %{
          type: "string",
          description: "问题的标题。参与者在看到问题时会看到的第一件事"
        },
        description: %{
          type: "string",
          description: "问题的内容。大约一段详细内容，定义了一个参与者要执行或回答的问题或任务"
        }
      }
    }
end

然后实现它：

<!-- tabs-open -->

同步

# 如果在路由器之外，您需要`require Phantom.Tool`。
# 如果在路由器中实现，这已经包含。
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("无效参数"), session}
  end
end

异步

# 如果在路由器之外，您需要`require Phantom.Tool`。
# 如果在路由器中实现，这已经包含。
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("无效参数"))
    end
  end)
  {:noreply, session}
end

<!-- tabs-close -->

定义提示

defmodule MyApp.MCP.Router do
  # ...

  # 提示可以包含参数。如果有参数，
  # 您可能还需要提供完成函数来帮助客户端填充参数。

  @description """
  审查提供的研究，并提供有关研究的有意义反馈，
  并告诉我是否有遗漏的问题。我们希望有一个能够提供研究目标见解的研究。
  """
  prompt :suggest_questions,
    completion_function: :study_complete,
    arguments: [
      %{
        name: "study_id",
        description: "要审查的研究",
        required: true
      }
    ]
end

然后实现它：

<!-- tabs-open -->

同步

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("哇塞"),
          assistant: Prompt.image(File.read!("foo.png")),
          user: Prompt.text("真的，哇塞")
        ), session}

    error ->
      {:error, Phantom.Request.internal_error(), session}
  end
end

异步

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("哇塞"),
          assistant: Prompt.image(File.read!("foo.png")),
          user: Prompt.text("真的，哇塞")
        ))

      error ->
        Session.respond(session, Phantom.Request.internal_error())
    end
  end)
  {:noreply, session}
end

<!-- tabs-close -->

定义资源

让我们定义一个带有资源模板的资源：

@description """
阅读研究的封面图片，以获得一些关于受众、研究目标和问题的背景信息。
"""
resource "myapp:///studies/:study_id/cover", :study_cover,
  completion_function: :study_complete,
  mime_type: "image/png"

@description """
阅读研究的内容。这包括问题和一般背景信息，有助于理解研究目标。
"""
resource "https://example.com/studies/:study_id/md", :study,
  completion_function: :study_complete,
  mime_type: "text/markdown"

然后实现它们：

<!-- tabs-open -->

同步

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

## 实现完成处理器：
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
    )

  # 您也可以返回一个包含更多信息的映射：
  # `%{values: study_ids, has_more: true, total: 1_000_000}`
  # 如果返回超过100个值，Phantom将设置`has_more: true`
  # 并只返回前100个。
  {:reply, study_ids, session}
end

异步

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

## 实现完成处理器：
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
    )

  # 您也可以返回一个包含更多信息的映射：
  # `%{values: study_ids, has_more: true, total: 1_000_000}`
  # 如果返回超过100个值，Phantom将设置`has_more: true`
  # 并只返回前100个。
  {:reply, study_ids, session}
end

<!-- tabs-close -->

您还应该在路由器中实现list_resources/2，以提供系统中所有可用资源的列表，并返回这些资源的链接。

@salt "cursor"
def list_resources(cursor, session) do
  # 记住要根据`session.allowed_resource_templates`检查允许的资源
  # 下面是一个玩具实现，仅供说明用途。
  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: "研究#{i}")
    end)

  {:reply,
    Resource.list(resource_links, next_cursor),
    session}
end

您可以通知客户端资源更新，如果他们订阅了任何资源的更新。

# 执行一些工作并更新底层资源，
# 然后通知任何监听者：
{:ok, uri} = MyApp.MCP.Router.resource_uri(:my_resource, id: "foo")
Phantom.Tracker.notify_resource_updated(uri)

PhantomMCP支持什么

Phantom将为您实现这些MCP请求：

• initialize。Phantom将根据在Phantom路由器中定义的工具检测客户端可用的能力。
• prompts/list 列出connect/2回调中提供的允许提示，或默认列出所有提示。要禁用，请在connect/2回调中返回allow_prompts(session, [])。
• prompts/get 将请求分派给您的处理程序，如果允许的话。更多内容请参阅Phantom.Prompt。
• resources/list 分派到您的MCP路由器。默认情况下，这是一个空列表，直到您实现它为止。更多内容请参阅Phantom.Resource。
• resource/templates/list 列出connect/2回调中提供的允许资源，或默认列出所有资源模板。要禁用，请在connect/2回调中返回allow_resource_templates(session, [])。更多内容请参阅Phantom.ResourceTemplate。
• resources/read 将请求分派给您的处理程序。请参阅Phantom.Resource。
• resources/subscribe 如果MCP路由器配置了pubsub，则可用。要通知资源更新，请使用Phantom.Tracker.notify_resource_updated(uri)。
• resources/unsubscribe 见上文。
• logging/setLevel 如果MCP路由器配置了pubsub，则可用。日志可以通过Session.log_{level}(session, map_content)发送到客户端。参见文档。
• tools/list 列出connect/2回调中提供的允许工具，或默认列出所有工具。要禁用，请在connect/2回调中返回allow_tools(session, [])。
• tools/call 将请求分派给您的处理程序。更多内容请参阅Phantom.Tool。
• completion/complete 将请求分派给给定提示或资源的完成处理器。
• notification/* 无操作。
• ping 响应pong。
• notifications/resources/list_changed - 服务器通知客户端资源列表已更新。这不是自动完成的；您需要使用Phantom.Tracker.notify_resource_list/0触发此事件，同时也要注意会话可能有权访问哪些资源。
• notifications/prompts/list_changed - 服务器通知客户端提示列表已更新。当调用Phantom.Cache.add_prompt/2时触发。
• notifications/tools/list_changed - 服务器通知客户端工具列表已更新。当调用Phantom.Cache.add_tool/2时触发。

Phantom目前不支持以下方法：

• roots/list - 服务器请求客户端提供可用于交互的文件列表。类似于resources/list，但针对客户端。
• sampling/createMessage - 服务器请求客户端查询其LLM并提供响应。这是人类参与的代理行动，可以在客户端请求服务器提示时利用。
• elicitation/create - 服务器请求客户端输入以完成客户端对其提出的请求。

批量请求

批量请求也将透明处理。请注意，没有抽象可以有效地将这些请求作为一个组提供给您的处理程序。由于MCP规范将在下一个版本中弃用批量请求支持，因此没有计划使其更高效。

身份验证和授权

Phantom本身不实现身份验证。需要身份验证的MCP应用程序应调查如Oidcc、Boruta或ExOauth2Provider等OAuth提供商解决方案，并配置路由以提供发现端点。

1. MCP身份验证和发现不由Phantom自身处理。您需要实现OAuth2并按照规范提供发现机制。在connect/2回调中，您可以返回{:unauthorized, www_authenticate_info}或{:forbidden, "错误消息"}以告知客户端如何继续。{:ok, session}结果意味着成功认证。

2. 一旦完成身份验证流程，对MCP路由器的请求应带有授权头，该头可以在您的MCP路由器的connect/2回调中接收和验证。

3. 您还可以根据授权规则限制可用的工具、提示或资源。下面是一个示例。

defmodule MyApp.MCP.Router do
  use Phantom.Router,
    name: "MyApp",