内容
# MCP over SSE
[](https://github.com/kEND/mcp_sse/releases)
[](https://hexdocs.pm/mcp_sse/)
[](https://modelcontextprotocol.io/specification/2024-11-05/index)
[](https://hex.pm/packages/mcp_sse)
[](https://github.com/kEND/mcp_sse/blob/main/LICENSE)
[](https://github.com/kEND/mcp_sse/actions/workflows/ci.yml?query=branch%3Amain)
[](https://github.com/kEND/mcp_sse/commits/main)
该库提供了一种简单的基于服务器端事件（SSE）的模型上下文协议（MCP）实现。
有关模型上下文协议的更多信息，请访问：
[模型上下文协议文档](https://modelcontextprotocol.io/introduction)。
## 目录
- [功能特性](#功能特性)
- [创建自己的MCP服务器](#创建自己的MCP服务器)
- [安装](#安装)
- [适用于Phoenix应用程序](#适用于Phoenix应用程序)
- [适用于使用Bandit的Plug应用程序](#适用于使用Bandit的Plug应用程序)
- [使用](#使用)
- [使用MCP检查器](#使用MCP检查器)
- [使用Cursor](#使用Cursor)
- [配置](#配置)
- [端口和HTTPS](#端口和HTTPS)
- [路径](#路径)
- [保持连接](#保持连接)
- [快速演示](#快速演示)
- [其他说明](#其他说明)
- [示例客户端用法](#示例客户端用法)
- [会话管理](#会话管理)
- [MCP响应格式化](#MCP响应格式化)
- [贡献](#贡献)
## 功能特性
- 完整的MCP服务器实现
- SSE连接管理
- JSON-RPC消息处理
- 工具注册和执行
- 会话管理
- 自动ping/保持连接
- 错误处理和验证
## 创建自己的MCP服务器
您必须实现`MCPServer`行为。
您只需要实现所需的回调函数（`handle_ping/1`和`handle_initialize/2`）以及您要支持的功能的任何可选回调函数。
`use MCPServer`宏提供：
- 内置消息路由
- 协议版本验证
- 可选回调的默认实现
- JSON-RPC错误处理
- 日志记录
参见[`DefaultServer`](https://hexdocs.pm/mcp_sse/MCPServer.html#module-example)以获取`MCPServer`行为的默认实现。
## 安装
### 适用于Phoenix应用程序：
1. 在`config/config.exs`中添加所需的配置：
```elixir
# 配置SSE的MIME类型
config :mime, :types, %{
"text/event-stream" => ["sse"]
}
# 配置MCP服务器
config :mcp_sse, :mcp_server, YourApp.YourMCPServer
```
2. 在`mix.exs`中添加依赖项：
```elixir
def deps do
[
{:mcp_sse, "~> 0.1.6"}
]
end
```
3. 配置路由器（`lib/your_app_web/router.ex`）：
```elixir
pipeline :sse do
plug :accepts, ["sse"]
end
scope "/" do
pipe_through :sse
get "/sse", SSE.ConnectionPlug, :call
post "/message", SSE.ConnectionPlug, :call
end
```
4. 运行应用程序：
```bash
mix phx.server
```
### 适用于使用Bandit的Plug应用程序：
1. 使用监督创建一个新的Plug应用程序：
```bash
mix new your_app --sup
```
2. 在`config/config.exs`中添加所需的配置：
```elixir
import Config
# 配置SSE的MIME类型
config :mime, :types, %{
"text/event-stream" => ["sse"]
}
# 配置MCP服务器
config :mcp_sse, :mcp_server, YourApp.YourMCPServer
```
3. 在`mix.exs`中添加依赖项：
```elixir
def deps do
[
{:mcp_sse, "~> 0.1.6"},
{:plug, "~> 1.14"},
{:bandit, "~> 1.2"}
]
end
```
4. 配置路由器（`lib/your_app/router.ex`）：
```elixir
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
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
```
5. 设置应用程序监督（`lib/your_app/application.ex`）：
```elixir
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. 运行应用程序：
```bash
mix run --no-halt
```
## 使用
### 使用MCP检查器
- 启动检查器：
```bash
MCP_SERVER_URL=localhost:4000 npx @modelcontextprotocol/inspector@latest
```
- 导航到 http://localhost:6274/
- 确保服务器正在运行
- 点击 `Connect`
- 现在您可以列出工具并调用它们
### 使用Cursor
- 打开Cursor设置
- 导航到MCP选项卡
- 点击 `Add new global MCP server`
- 使用以下内容填充 `~/.cursor/mcp.json`：
```json
{
"mcpServers": {
"your-mcp-server": {
"url": "http://localhost:4000/sse"
}
}
}
```
- 确保服务器正在运行
- 请求Cursor运行其中一个工具
## 配置
### 端口和HTTPS
Bandit服务器可以使用应用程序模块中的其他选项进行配置：
```elixir
# 示例，使用自定义端口和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和消息端点的路径：
```elixir
config :mcp_sse,
sse_path: "/mcp/sse", # 默认："/sse"
message_path: "/mcp/msg" # 默认："/message"
```
这允许您在路由器中使用自定义路径：
```elixir
# 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间隔或完全禁用它：
```elixir
# 设置自定义ping间隔（毫秒）
config :mcp_sse, :sse_keepalive_timeout, 30_000 # 30秒
# 或完全禁用ping
config :mcp_sse, :sse_keepalive_timeout, :infinity
```
## 快速演示
要查看MCP服务器的运行情况：
1. 在一个终端中启动服务器：
```bash
# 我们的示例服务器
elixir dev/example_server.exs
# 您的Phoenix应用程序
mix phx.server
# 您的Plug应用程序
mix run --no-halt
```
2. 在另一个终端中，运行演示客户端脚本：
```bash
elixir dev/example_client.exs
```
客户端脚本将：
- 连接到SSE端点
- 初始化连接
- 列出可用工具
- 调用具有示例输入的upcase工具
- 显示每个步骤的结果
这提供了一个实际演示模型上下文协议流程和服务器功能。
## 其他说明
### 示例客户端用法
```javascript
// 连接到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规范的内容类型。
响应内容应格式化为以下类型之一：
```elixir
# 文本内容
{: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: "resource description"
}
}
]
}
}}
```
对于JSON等结构化数据，您应将其转换为格式化字符串：
```elixir
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内容类型规范](https://spec.modelcontextprotocol.io/specification/2024-11-05/basic/messages/#responses)。
## 贡献
- 复刻仓库并克隆它
- 在您的复刻中创建一个新分支
- 进行更改并提交它们
- 将更改推送到您的复刻
- 在上游打开拉取请求
连接信息
猜你喜欢
暂无推荐服务器