MCP SSE Proxy Server

MCP SSE 代理服务

MCP SSE 代理服务是一个基于服务器发送事件（SSE）的模型上下文协议（MCP）代理服务，可通过标准 MCP 协议远程执行 MCP 服务。它支持共享和独立两种会话模式。

模型上下文协议（MCP）是一种开放协议，用于标准化应用向大模型提供上下文的方式。可以把 MCP 想象成 AI 应用的 USB-C 接口。正如 USB-C 提供了一种标准化方式，让设备连接到各种外围设备和配件，MCP 也提供了一种标准化方式，让 AI 模型连接到不同的数据源和工具。

如需了解更多 MCP 信息：

• MCP 文档（英文）
• MCP 文档（中文）
• MCP 服务目录 - 发现并探索各种 MCP 服务

中文文档

它解决了哪些问题？

本项目可帮助你：

• 使用标准 MCP 协议远程启动并与任何 MCP 服务交互
• 通过 SSE 保持稳定的长连接
• 支持多个客户端共享一个 MCP 会话，或各自维护独立会话
• 通过请求参数动态配置服务环境
• 通过 NPX 和 UVX 快速部署 STDIO 进程

系统架构

功能

• 支持 SSE 长连接以推送服务端消息
• 支持 JSON-RPC 2.0 消息格式
• 支持共享和独立两种会话模式
• 通过请求参数动态配置会话环境
• 自动保持连接（30 秒心跳）
• 完善的错误处理和日志记录
• 内置支持部署 NPX 和 UVX STDIO 进程

集成与自定义

自定义 Docker 镜像

MCP SSE 代理服务旨在轻松集成到各种自定义 Docker 镜像中。这种灵活性使你能够针对不同使用场景创建专用环境：

• 浏览器自动化：集成 Chrome/Chromium，以支持基于浏览器的 MCP 服务
• 数据库操作：与特定数据库打包，以支持数据处理 MCP 服务
• 开发工具：与开发工具一同打包，以支持代码相关的 MCP 服务
• AI/ML 工具：包含 AI/ML 库，以支持机器学习 MCP 服务

高级路由

默认配置通过环境变量加载 MCP 服务命令，但你可以通过以下方式扩展功能：

1. 多服务集成：通过配置文件配置多个 MCP 服务
2. 动态路由：在 SSE 连接中添加路由字段，将流量导向不同的 MCP 服务
3. 自定义路由器：扩展此基础实现，构建你自己的 MCP 路由器项目

这种灵活性使你能够：

• 创建专用的 MCP 服务集群
• 在多个服务之间实现负载均衡
• 根据需求设计自定义路由逻辑
• 构建分层 MCP 服务架构

配置

环境变量

• SHARED_SESSION：控制会话模式

  • true：共享会话模式（默认），所有客户端共享一个 MCP 会话
  • false：独立会话模式，每个客户端创建单独的 MCP 会话

• AUTH_KEY：服务访问密钥

  • 如果已设置，所有请求都必须提供此密钥
  • 通过 URL 参数 auth_key 传递

• MCP_SERVER_CONFIG：MCP 服务配置 示例：

  # Run Node.js-based MCP server using NPX
  MCP_SERVER_CONFIG="npx -y @modelcontextprotocol/server-filesystem ."
  
  # Run Python-based MCP server using UVX
  MCP_SERVER_CONFIG="uvx mcp-server-fetch"
  

动态配置

在独立会话模式（SHARED_SESSION=false）下，每个 SSE 连接都会启动一个新的 MCP 服务进程。你可以通过 URL 参数为每个会话动态配置环境变量：

GET /sse?auth_key=xxx&[CUSTOM_ENV]=value

这种机制对于需要用户专属凭据的场景特别有用。例如，部署 GitHub MCP 服务时，不同用户可以使用各自的 Personal Access Token：

1. 部署服务时配置基础命令：

export MCP_SERVER_CONFIG="npx -y @modelcontextprotocol/server-github"

2. 用户连接时通过 URL 提供自己的令牌：

GET /sse?auth_key=xxx&GITHUB_PERSONAL_ACCESS_TOKEN=ghp_xxxxxxxxxxxx

这样，你只需部署一个 MCP 代理服务即可为多个用户提供服务，每个用户都可以使用自己的 GitHub 凭据。其他支持的环境变量配置：

• 通用配置：

  • NODE_ENV：Node.js 环境（development/production）
  • DEBUG：调试日志级别

• 特定服务配置：

  • GitHub MCP 服务：GITHUB_PERSONAL_ACCESS_TOKEN
  • Filesystem MCP 服务：ROOT_DIR
  • 其他特定服务的环境变量

注意：URL 参数中的环境变量会覆盖服务的默认配置。

API 端点

SSE 连接

GET /sse?auth_key=xxx

成功建立连接后，服务会返回一个消息端点 URL：

event: endpoint
data: /messages?session_id=<session_id>

发送消息

POST /messages?session_id=<session_id>
Content-Type: application/json

{
    "jsonrpc": "2.0",
    "method": "method_name",
    "params": {},
    "id": 1
}

支持的方法

• initialize：初始化会话
• tools/list：列出可用工具
• tools/call：调用工具
• prompts/list：列出可用提示词
• prompts/get：获取提示词
• resources/list：列出资源
• resources/templates/list：列出资源模板
• resources/read：读取资源
• resources/subscribe：订阅资源
• resources/unsubscribe：取消订阅资源

运行服务

使用 Python

# Set environment variables
export AUTH_KEY=your_key
export SHARED_SESSION=true  # or false

# Start the server
python main.py

服务默认在 0.0.0.0:8000 上启动。

使用 Docker

1. 拉取 Docker 镜像：

docker pull codefriday123/mcpproxy:1.0

2. 运行容器：

docker run -d \
  -p 8000:8000 \
  -e AUTH_KEY=your_key \
  -e SHARED_SESSION=true \
  -e MCP_SERVER_CONFIG="npx -y @modelcontextprotocol/server-filesystem ." \
  codefriday123/mcpproxy:1.0

环境变量：

• AUTH_KEY：服务访问密钥（必需）
• SHARED_SESSION：会话模式（默认：true）
• MCP_SERVER_CONFIG：MCP 服务配置命令

服务将在 http://localhost:8000 上提供访问。

使用 Alipay Cloud Run

你也可以在 Alipay Cloud Run 上部署服务（新人可免费使用容器资源）：

1. 在 Cloud Run 中创建新服务

2. 使用以下设置配置服务：

   • 镜像：codefriday123/mcpproxy:1.0
   • 端口：8000
   • 环境变量：
     • AUTH_KEY：你的服务访问密钥
     • SHARED_SESSION：true/false
     • MCP_SERVER_CONFIG：你的 MCP 服务配置

3. 部署服务

服务将在提供的 Cloud Run 端点上开放访问。

日志记录

服务使用 Python 的 logging 模块记录详细日志，包括：

• 环境变量信息
• 连接的建立和关闭
• 消息处理
• 错误和异常

错误处理

服务使用标准的 JSON-RPC 2.0 错误响应格式：

{
    "jsonrpc": "2.0",
    "error": {
        "code": error_code,
        "message": "error message"
    },
    "id": null
}

主要错误代码：

• -32700：解析错误
• -32600：无效请求
• -32601：未找到方法
• -32602：参数无效
• -32603：内部错误
• -32000：服务错误

使用示例

Python 客户端

你可以使用官方 MCP Python 客户端连接已部署的 MCP 服务：

import asyncio
from typing import Optional
from contextlib import AsyncExitStack
from mcp import ClientSession, LoggingLevel
from mcp.client.sse import sse_client

class MCPClient:
    def __init__(self):
        self.session: Optional[ClientSession] = None
        self._streams_context = None
        self._session_context = None
        self.exit_stack = AsyncExitStack()

    async def connect_to_sse_server(self, server_url: str):
        """Connect to an MCP server running SSE transport"""
        try:
            print(f"Connecting to {server_url}...")
            self._streams_context = sse_client(url=server_url)
            streams = await self._streams_context.__aenter__()
            self._session_context = ClientSession(*streams)
            self.session = await self._session_context.__aenter__()
            
            try:
                await asyncio.wait_for(self.session.initialize(), timeout=30.0)
                print("Session initialized successfully")
            except asyncio.TimeoutError:
                print("Session initialization timed out, but will continue")

        except Exception as e:
            print(f"Connection error: {e}")
            await self.cleanup()
            raise

    async def cleanup(self):
        """Properly clean up session and streams"""
        try:
            if hasattr(self, '_session_context') and self._session_context:
                await self._session_context.__aexit__(None, None, None)
            if hasattr(self, '_streams_context') and self._streams_context:
                await self._streams_context.__aexit__(None, None, None)
        except Exception as e:
            print(f"Cleanup error: {e}")

async def main():
    client = MCPClient()
    try:
        await client.connect_to_sse_server('http://your-server:8000/sse?auth_key=your_auth_key')
        response = await asyncio.wait_for(client.session.list_tools(), timeout=10.0)
        print(f"Available tools: {[tool.name for tool in response.tools]}")
    except Exception as e:
        print(f"Error: {e}")
    finally:
        await client.cleanup()

if __name__ == "__main__":
    asyncio.run(main())

### Cursor Integration

You can also integrate with [Cursor](https://docs.cursor.com/context/model-context-protocol) by adding the server to Cursor's MCP configuration:

```json
{
  "mcpServers": {
    "server-name": {
      "url": "http://your-server:8000/sse?auth_key=your_auth_key",
      "env": {
      }
    }
  }
}

将此配置放在以下任一位置：

• 项目目录中的 .cursor/mcp.json，用于项目专属访问
• 主目录中的 ~/.cursor/mcp.json，用于全局访问

许可证

MIT 许可证

贡献

欢迎贡献！欢迎随时提交 Pull Request。

路线图

当前功能

• 基本 MCP 服务代理功能
• 支持 SSE 传输
• 共享和独立会话模式
• 动态环境配置
• 集成 Cursor 和其他 MCP 客户端

未来计划

• 无状态架构

  • 用于无状态运行的纯工具模式
  • 无需会话管理的基于 HTTP 的工具执行
  • 简化部署与扩展
  • 降低资源开销
  • 提高可靠性和可维护性

• Agent 沙箱

  • 创建隔离环境以运行 MCP Agent
  • 资源限制与监控
  • 为不受信任的 Agent 提供安全沙箱
  • 优化 Agent 执行性能
  • 支持可流式传输的 HTTP
    • 与可流式传输的 HTTP 协议保持向前兼容
    • 混合模式：SSE + HTTP，提升灵活性
    • 协议协商与回退机制
    • 优化流式传输性能

• Agent 路由器

  • 在多个 MCP 服务之间进行智能路由
  • 负载均衡与故障转移
  • 请求／响应转换
  • 处理协议版本兼容性
  • 高级路由规则与策略

• 最佳实践

  • 针对不同场景的部署指南
  • 安全建议
  • 性能优化技巧
  • 监控与日志记录最佳实践
  • 与各种 MCP 服务集成的模式

敬请关注更新与新功能！