星战科技xingzhan.cloud
返回市场
MCP服务器sse

MCP服务器sse

作者：kEND64 星标更新：2025-05-04
安装使用来源：spider==github.com

项目介绍

MCP over SSE

发布 文档 MCP规范版本 下载量 许可证 持续集成 最近一次提交

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

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

目录

特性

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

创建自己的MCP服务器

您必须实现MCPServer行为。

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

use MCPServer宏提供了：

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

参见DefaultServer以获取MCPServer行为的默认实现。

安装

对于Phoenix应用程序：

  1. 在config/config.exs中添加所需配置：
# 配置SSE的MIME类型
config :mime, :types, %{
  "text/event-stream" => ["sse"]
}

# 配置MCP服务器
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的Plug应用程序：

  1. 使用监督创建一个新的Plug应用程序：
mix new your_app --sup
  1. 在config/config.exs中添加所需配置：
import Config

# 配置SSE的MIME类型
config :mime, :types, %{
  "text/event-stream" => ["sse"]
}

# 配置MCP服务器
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

  # 中间件确保会话ID存在
  def ensure_session_id(conn, _opts) do
    case get_session_id(conn) do
      nil ->
        # 如果没有会话ID，则生成新的会话ID
        session_id = generate_session_id()
        %{conn | query_params: Map.put(conn.query_params, "sessionId", session_id)}
      _session_id ->
        conn
    end
  end

  # 辅助函数从查询参数中获取会话ID
  defp get_session_id(conn) do
    conn.query_params["sessionId"]
  end

  # 生成唯一的会话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 Inspector一起使用

  • 启动检查器：
MCP_SERVER_URL=localhost:4000 npx @modelcontextprotocol/inspector@latest
  • 导航到 http://localhost:6274/
  • 确保您的服务器正在运行
  • 点击“连接”
  • 您现在可以列出工具并调用它们

与Cursor一起使用

  • 打开Cursor设置
  • 导航到MCP标签页
  • 点击“添加新的全局MCP服务器”
  • 填写~/.cursor/mcp.json：
{
  "mcpServers": {
    "your-mcp-server": {
      "url": "http://localhost:4000/sse"
    }
  }
}
  • 确保您的服务器正在运行
  • 要求Cursor运行您的一个工具

配置

端口和HTTPS

Bandit服务器可以在您的应用程序模块中通过额外选项进行配置：

# 示例：自定义端口和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",    # 默认值："/sse"
  message_path: "/mcp/msg" # 默认值："/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间隔或完全禁用它：

# 设置自定义ping间隔（以毫秒为单位）
config :mcp_sse, :sse_keepalive_timeout, 30_000  # 30秒

# 或者完全禁用ping
config :mcp_sse, :sse_keepalive_timeout, :infinity

快速演示

要查看MCP服务器的实际操作：

  1. 在一个终端中启动服务器：
# 我们的示例服务器
elixir dev/example_server.exs

# 您的Phoenix应用程序
mix phx.server

# 您的Plug应用程序
mix run --no-halt
  1. 在另一个终端中运行演示客户端脚本：
elixir dev/example_client.exs

客户端脚本将：

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

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

其他注意事项

示例客户端用法

// 连接到SSE端点
const sse = new EventSource('/sse');

// 处理端点消息
sse.addEventListener('endpoint', (e) => {
  const messageEndpoint = e.data;
  // 使用messageEndpoint进行后续的JSON-RPC请求
});

// 发送初始化请求
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规范的内容类型。 响应内容应格式化为以下类型之一：

# 文本内容
{:ok,
 %{
   jsonrpc: "2.0",
   id: request_id,
   result: %{
     content: [
       %{
         type: "text",
         text: "您的文本响应在这里"
       }
     ]
   }
 }}

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

# 资源引用
{:ok,
 %{
   jsonrpc: "2.0",
   id: request_id,
   result: %{
     content: [
       %{
         type: "resource",
         resource: %{
           name: "resource_name",
           description: "资源描述"
         }
       }
     ]
   }
 }}

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

def handle_call_tool(request_id, %{"name" => "list_companies"} = _params) do
  companies = fetch_companies()  # 您的数据获取逻辑

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

有关响应格式化的更多细节，请参阅MCP内容类型规范。

贡献

  • 分叉仓库并克隆它
  • 在您的分叉中创建一个新分支
  • 提交您的更改
  • 将更改推送到您的分叉
  • 在上游打开一个拉取请求