MCP over SSE

基于 SSE 的 MCP

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

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

目录

• 功能
• 创建自己的 MCP 服务
• 安装
  • 用于 Phoenix 应用
  • 用于搭配 Bandit 的 Plug 应用
• 使用方法
  • 搭配 MCP Inspector 使用
  • 搭配 Cursor 使用
• 配置
  • 端口和 HTTPS
  • 路径
  • 保活
• 快速演示
• 其他说明
  • 客户端使用示例
  • 会话管理
  • MCP 响应格式
• 贡献

功能

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

创建自己的 MCP 服务

你必须实现 MCPServer behaviour。

你只需实现必需的回调（handle_ping/1 和 handle_initialize/2），以及你希望支持的功能对应的可选回调。

use MCPServer 宏提供：

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

请参阅 DefaultServer，了解 MCPServer behaviour 的默认实现。

安装

用于 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

2. 在 mix.exs 中添加依赖：

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

3. 配置路由器（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

4. 运行应用：

mix phx.server

用于搭配 Bandit 的 Plug 应用：

1. 创建一个带监督机制的新 Plug 应用：

mix new your_app --sup

2. 在 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

3. 在 mix.exs 中添加依赖：

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

4. 配置路由器（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

5. 设置应用监督（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

6. 运行应用：

mix run --no-halt

使用方法

搭配 MCP Inspector 使用

• 启动 Inspector：

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

• 打开 http://localhost:6274/
• 确保你的服务正在运行
• 点击 Connect
• 现在可以列出工具并调用它们

搭配 Cursor 使用

• 打开 Cursor 设置
• 前往 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，以防止连接超时。 你可以在 config/config.exs 中配置 ping 间隔，或将其完全禁用：

# 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

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

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，则生成新的会话 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 此仓库并克隆它
• 在你的 fork 中创建一个新分支
• 进行更改并提交
• 将更改推送到你的 fork
• 在上游仓库中发起 pull request