SSE MCP Template

SSE MCP服务器模板

一个生产就绪的、受领域驱动设计（DDD）启发的模板，用于使用Python、FastAPI和服务器发送事件（SSE）构建模型上下文协议（MCP）服务器。

特性

• 生产就绪：结构化以实现可扩展性和可维护性
• 苏格兰和南方能源公司运输：MCP的服务器发送事件传输的本机支持
• 快速API：基于高性能FastAPI构建
• 类型安全：严格的类型检查 mypy 以及全面的类型提示
• DDD架构：域、应用程序、基础架构和表示层的清晰分离
• 测试：已配置 pytest 和 pytest-asyncio

项目结构

src/sse_mcp_server/
├── domain/           # Entities, Value Objects, Protocols
│   ├── models.py     # Domain models and exceptions
│   └── protocols.py  # Domain interfaces
├── application/      # Application Services, Use Cases
│   └── services.py   # Business logic implementation
├── infrastructure/   # External implementations (MCP Server, adapters)
│   └── mcp_server.py # MCP server configuration
├── presentation/     # API Endpoints (SSE, HTTP)
│   └── api.py        # FastAPI routes and SSE handlers
└── main.py           # Application entry point

先决条件

• Python 3.10+
• pip 或 uv

入门指南

1.安装

# Clone the template
git clone git@github.com:M1T8E6/SSE-MCP-Template.git
cd sse-mcp-server 

# Open the project in the dev-container or set up a virtual environment
uv venv venv
source venv/bin/activate

# Install dependencies
uv pip install -r requirements.txt

2.配置

make prepare-env

3.运行服务器

make run

可用端点：

• 健康检查： http://0.0.0.0:5001/mcp/v1/health
• SSE连接： http://0.0.0.0:5001/mcp/v1/sse
• MCP消息： http://0.0.0.0:5001/mcp/v1/messages （职位）

4.开发工作流程

# Run tests
make test

# Lint and type check
make lint

# Clean cache and build files
make clean

建筑细部

域层

包含纯业务逻辑和接口（协议）。它不依赖于外层。

• models.py：域实体、值对象和异常
• protocols.py：域服务接口

应用层

使用域协议实现用例和应用程序服务。

• 服务.py：应用程序服务实现

基础设施层

实现域中定义的协议并提供外部集成。

• mcp_server.py：MCP服务器配置，包括工具、资源和提示

表示层

处理HTTP请求和SSE连接。将网络世界与应用程序服务连接起来。

• api.py：FastAPI路线和SSE运输处理

MCP集成

此模板使用低级 mcp.server.Server 以允许与FastAPI完全集成。

• 这 SseServerTransport 与FastAPI路由正确集成
• 工具和资源定义见 infrastructure/mcp_server.py
• SSE端点建立连接并管理MCP服务器生命周期
• 消息被发布到单独的端点进行处理

示例：添加新工具

编辑 src/sse_mcp_server/infrastructure/mcp_server.py:

@mcp_server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="your_tool",
            description="Description of your tool",
            inputSchema={
                "type": "object",
                "properties": {
                    "param": {"type": "string"},
                },
                "required": ["param"],
            },
        )
    ]

@mcp_server.call_tool()
async def handle_call_tool(
    name: str, arguments: dict[str, object] | None
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    if name == "your_tool":
        param = arguments.get("param") if arguments else None
        result = f"Tool executed with: {param}"
        return [types.TextContent(type="text", text=result)]
    raise ValueError(f"Unknown tool: {name}")

测试

使用以下工具运行测试：

make test

许可证

麻省理工学院