MCP通过SSE
  ](https://modelcontextprotocol.io/specification/2024-11-05/index) ](https://hex.pm/packages/mcp_sse)   
此库提供了服务器发送事件（SSE）上模型上下文协议（MCP）的简单实现。
有关模型上下文协议的更多信息，请访问： 模型上下文协议文档.
目录
- 适用于Phoenix应用 - 适用于Bandit的插头应用
特性
- 完整的MCP服务器实施
- SSE连接管理
- JSON-RPC消息处理
- 工具注册和执行
- 会话管理
- 自动ping/keepalive
- 错误处理和验证
创建自己的MCP服务器
您必须执行 MCPServer 行为。
您只需要实现所需的回调(handle_ping/1 和 handle_initialize/2)以及您想要支持的功能的任何可选回调。
这 use MCPServer 宏提供：
- 内置消息路由
- 协议版本验证
- 可选回调的默认实现
- JSON-RPC错误处理
- 日志记录
看 DefaultServer 对于的默认实现 MCPServer 行为。
安装
对于Phoenix应用：
- 将所需配置添加到
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- 添加到您的依赖项中
mix.exs:
def deps do
[
{:mcp_sse, "~> 0.1.6"}
]
end- 配置路由器(
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- 运行您的应用程序：
mix phx.server对于使用Bandit的插头应用：
- 在监督下创建新的Plug应用程序：
mix new your_app --sup- 将所需配置添加到
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- 向添加依赖项
mix.exs:
def deps do
[
{:mcp_sse, "~> 0.1.6"},
{:plug, "~> 1.14"},
{:bandit, "~> 1.2"}
]
end- 配置路由器(
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- 设置您的应用程序监督(
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- 运行您的应用程序：
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服务器的运行情况：
- 在一个终端中启动服务器：
# Our example server
elixir dev/example_server.exs
# Your Phoenix application
mix phx.server
# Your Plug application
mix run --no-halt- 在另一个终端中，运行演示客户端脚本：
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中创建一个新分支
- 进行更改并提交
- 将更改推到叉子上
- 在上游打开拉取请求