MCP over SSE

基本能力

产品定位

mcp_sse 是一个基于 Server-Sent Events (SSE) 实现的 Model Context Protocol (MCP) 服务器库，旨在提供实时通信和协议处理能力。

核心功能

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

适用场景

• 需要实时通信的应用程序
• 基于 Phoenix 或 Plug 的应用
• 需要协议处理的系统

工具列表

• SSE.ConnectionPlug: 处理 SSE 连接和消息路由
• MCP.DefaultServer: 默认的 MCP 服务器实现
• YourApp.YourMCPServer: 自定义的 MCP 服务器实现

常见问题解答

• 如何配置 SSE 连接的 keepalive 间隔？ 在 config/config.exs 中设置 :sse_keepalive_timeout 参数。
• 如何禁用 keepalive？ 在 config/config.exs 中设置 :sse_keepalive_timeout 为 :infinity。
• 如何生成 session ID？ 路由器会自动生成 session ID 如果不存在。

使用教程

使用依赖

对于 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, MCP.DefaultServer
# config :mcp_sse, :mcp_server, YourApp.YourMCPServer

1. 添加依赖到 mix.exs:

def deps do
  [
    {:mcp_sse, "~> 0.1.0"}
  ]
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

  pipe_through :api
  post "/message", SSE.ConnectionPlug, :call
end

对于 Plug 应用 (使用 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.MCPServer

1. 添加依赖到 mix.exs:

def deps do
  [
    {:mcp_sse, "~> 0.1.0"},
    {: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: Jason

  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. 按照上述依赖部分配置你的应用。
2. 运行 mix deps.get 获取依赖。
3. 启动你的应用。

调试方式

1. 启动 Phoenix 服务器:

mix phx.server

1. 运行示例客户端脚本:

elixir examples/mcp_client.exs

1. 观察客户端脚本的输出，确保连接和消息处理正常。

许可证

该项目遵循 MIT 开源许可条款，请参阅 MIT 了解完整条款。