MCP SSE

MCP通过SSE

此库提供了服务器发送事件（SSE）上模型上下文协议（MCP）的简单实现。

有关模型上下文协议的更多信息，请访问： 模型上下文协议文档.

目录

• 特性
• 创建自己的MCP服务器
• 安装

- 适用于Phoenix应用 - 适用于Bandit的插头应用

• 用法

- 与MCP检查员一起 - 带光标

• 配置

- 端口和HTTPS - 路径 - 存活

• 快速演示
• 其他注意事项

- 客户端使用示例 - 会话管理 - MCP响应格式

• 贡献

特性

• 完整的MCP服务器实施
• SSE连接管理
• JSON-RPC消息处理
• 工具注册和执行
• 会话管理
• 自动ping/keepalive
• 错误处理和验证

创建自己的MCP服务器

您必须执行 MCPServer 行为。

您只需要实现所需的回调(handle_ping/1 和 handle_initialize/2)以及您想要支持的功能的任何可选回调。

这 use MCPServer 宏提供：

• 内置消息路由
• 协议版本验证
• 可选回调的默认实现
• JSON-RPC错误处理
• 日志记录

看 DefaultServer 对于的默认实现 MCPServer 行为。

安装

对于Phoenix应用：

1. 将所需配置添加到 config/config.exs:

# Configure MIME types for SSE
config :mime, :types, %{
  "text/event-stream" => ["sse"]
}

# Configure the MCP Server
config :mcp_sse, :mcp_server, YourApp.YourMCPServer

1. 添加到您的依赖项中 mix.exs:

def deps do
  [
    {:mcp_sse, "~> 0.1.6"}
  ]
end

1. 配置路由器(lib/your_app_web/router.ex):

pipeline :sse do
  plug :accepts, ["sse"]
end

scope "/" do
  pipe_through :sse
  get "/sse", SSE.ConnectionPlug, :call
  post "/message", SSE.ConnectionPlug, :call
end

1. 运行您的应用程序：

mix phx.server

对于使用Bandit的插头应用：

1. 在监督下创建新的Plug应用程序：

mix new your_app --sup

1. 将所需配置添加到 config/config.exs:

import Config

# Configure MIME types for SSE
config :mime, :types, %{
  "text/event-stream" => ["sse"]
}

# Configure the MCP Server
config :mcp_sse, :mcp_server, YourApp.YourMCPServer

1. 向添加依赖项 mix.exs:

def deps do
  [
    {:mcp_sse, "~> 0.1.6"},
    {:plug, "~> 1.14"},
    {:bandit, "~> 1.2"}
  ]
end

1. 配置路由器(lib/your_app/router.ex):

defmodule YourApp.Router do
  use Plug.Router

  plug Plug.Parsers,
    parsers: [:urlencoded, :json],
    pass: ["text/*"],
    json_decoder: JSON

  plug :match
  plug :ensure_session_id
  plug :dispatch

  # Middleware to ensure session ID exists
  def ensure_session_id(conn, _opts) do
    case get_session_id(conn) do
      nil ->
        # Generate a new session ID if none exists
        session_id = generate_session_id()
        %{conn | query_params: Map.put(conn.query_params, "sessionId", session_id)}
      _session_id ->
        conn
    end
  end

  # Helper to get session ID from query params
  defp get_session_id(conn) do
    conn.query_params["sessionId"]
  end

  # Generate a unique session ID
  defp generate_session_id do
    Base.encode16(:crypto.strong_rand_bytes(8), case: :lower)
  end

  forward "/sse", to: SSE.ConnectionPlug
  forward "/message", to: SSE.ConnectionPlug

  match _ do
    send_resp(conn, 404, "Not found")
  end
end

1. 设置您的应用程序监督(lib/your_app/application.ex):

defmodule YourApp.Application do
  use Application

  @impl true
  def start(_type, _args) do
    children = [
      {Bandit, plug: YourApp.Router, port: 4000}
    ]

    opts = [strategy: :one_for_one, name: YourApp.Supervisor]
    Supervisor.start_link(children, opts)
  end
end

1. 运行您的应用程序：

mix run --no-halt

用法

与MCP检查员一起

• 启动检查器：

MCP_SERVER_URL=localhost:4000 npx @modelcontextprotocol/inspector@latest

• 引导到http://localhost:6274/
• 确保您的服务器正在运行
• 点击 Connect
• 现在，您可以列出工具并调用它们

带光标

• 打开光标设置
• 导航到MCP选项卡
• 点击 Add new global MCP server
• 填写 ~/.cursor/mcp.json 与：

{
  "mcpServers": {
    "your-mcp-server": {
      "url": "http://localhost:4000/sse"
    }
  }
}

• 确保您的服务器正在运行
• 让Cursor运行您的工具之一

配置

端口和HTTPS

Bandit服务器可以在应用程序模块中配置其他选项：

# Example with custom port and HTTPS
children = [
  {Bandit,
    plug: YourApp.Router,
    port: System.get_env("PORT", "4000") |> String.to_integer(),
    scheme: :https,
    certfile: "priv/cert/selfsigned.pem",
    keyfile: "priv/cert/selfsigned_key.pem"
  }
]

路径

您可以自定义用于SSE和消息端点的路径：

config :mcp_sse,
  sse_path: "/mcp/sse",    # Default: "/sse"
  message_path: "/mcp/msg" # Default: "/message"

这允许您在路由器中使用自定义路径：

# Phoenix
scope "/mcp" do
  pipe_through :sse
  get "/sse", SSE.ConnectionPlug, :call
  post "/msg", SSE.ConnectionPlug, :call
end

# Plug
forward "/mcp/sse", to: SSE.ConnectionPlug
forward "/mcp/msg", to: SSE.ConnectionPlug

存活

SSE连接定期发送保活ping，以防止连接超时。 您可以在中配置ping间隔或完全禁用它 config/config.exs:

# Set custom ping interval (in milliseconds)
config :mcp_sse, :sse_keepalive_timeout, 30_000  # 30 seconds

# Or disable pings entirely
config :mcp_sse, :sse_keepalive_timeout, :infinity

快速演示

要查看MCP服务器的运行情况：

1. 在一个终端中启动服务器：

# Our example server
elixir dev/example_server.exs

# Your Phoenix application
mix phx.server

# Your Plug application
mix run --no-halt

1. 在另一个终端中，运行演示客户端脚本：

elixir dev/example_client.exs

客户端脚本将：

• 连接到SSE端点
• 初始化连接
• 列出可用工具
• 使用示例输入调用upcase工具
• 显示每个步骤的结果

这提供了模型上下文协议流和服务器功能的实际演示。

其他注意事项

客户端使用示例

// Connect to SSE endpoint
const sse = new EventSource('/sse');

// Handle endpoint message
sse.addEventListener('endpoint', (e) => {
  const messageEndpoint = e.data;
  // Use messageEndpoint for subsequent JSON-RPC requests
});

// Send initialize request
fetch('/message?sessionId=YOUR_SESSION_ID', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({
    jsonrpc: '2.0',
    id: 1,
    method: 'initialize',
    params: {
      protocolVersion: '2024-11-05',
      capabilities: {}
    }
  })
});

会话管理

MCP SSE服务器要求每个连接都有一个会话ID。路由器会自动执行以下操作：

• 使用查询参数中的现有会话ID（如果提供）
• 如果不存在，则生成新的会话ID
• 确保所有请求 /sse 和 /message 终结点具有有效的会话ID

MCP响应格式

在MCP服务器中实现工具响应时，内容必须遵循MCP内容类型规范。 响应内容的格式应为以下类型之一：

# Text content
{:ok,
 %{
   jsonrpc: "2.0",
   id: request_id,
   result: %{
     content: [
       %{
         type: "text",
         text: "Your text response here"
       }
     ]
   }
 }}

# Image content
{:ok,
 %{
   jsonrpc: "2.0",
   id: request_id,
   result: %{
     content: [
       %{
         type: "image",
         data: "base64_encoded_image_data",
         mimeType: "image/png"
       }
     ]
   }
 }}

# Resource reference
{:ok,
 %{
   jsonrpc: "2.0",
   id: request_id,
   result: %{
     content: [
       %{
         type: "resource",
         resource: %{
           name: "resource_name",
           description: "resource description"
         }
       }
     ]
   }
 }}

对于JSON等结构化数据，您应该将其转换为格式化字符串：

def handle_call_tool(request_id, %{"name" => "list_companies"} = _params) do
  companies = fetch_companies()  # Your data fetching logic

  {:ok,
   %{
     jsonrpc: "2.0",
     id: request_id,
     result: %{
       content: [
         %{
           type: "text",
           text: JSON.encode!(companies, pretty: true)
         }
       ]
     }
   }}
end

有关响应格式的更多详细信息，请参阅 MCP内容类型规范.

贡献

• 分叉存储库并克隆它
• 在fork中创建一个新分支
• 进行更改并提交
• 将更改推到叉子上
• 在上游打开拉取请求