MCP Uv Base

FastMCP服务器

使用FastMCP构建的模型上下文协议（MCP）服务器，演示了可流式传输的HTTP传输和服务器发送事件（SSE）功能。

特性

• ✨ 流式HTTP传输 -支持SSE的基于HTTP的MCP服务器
• 🔧 多种工具 -添加整数、问候用户和流式传输实时数据
• 🚀 实时流媒体 -发电机具有基于时间的处理功能
• ⚡ 无状态模式 -横向可扩展，无会话管理
• 🐳 开发容器就绪 -完整的开发环境

快速开始

地方发展

# Install dependencies
uv sync

# Start the FastMCP server
uv run python main.py

# Server runs on http://localhost:3001
# MCP endpoint: http://localhost:3001/mcp
# SSE endpoint: http://localhost:3001/sse

使用开发容器

此存储库包括一个完整的开发容器设置，用于跨环境进行一致的开发。

先决条件

• VS Code
• 码头工人
• 开发容器扩展

快速设置

1. 克隆此存储库
2. 在VS代码中打开
3. 出现提示时，单击“在容器中重新打开”（或使用命令面板：“开发容器：在容器中再次打开”）
4. 等待容器构建和依赖项安装
5. 运行服务器： uv run python main.py

dev容器包括：

• 带uv包管理器的Python 3.12
• 预配置的VS代码扩展（Python、Pylance等）
• Zsh与Oh My Zsh
• Git和GitHub CLI
• FastMCP服务器的端口转发（3001、8000、8080）

可用工具

add(a: int, b: int) -> int

将两个整数相加。

greet(name: str) -> str

返回友好的问候信息。

stream_numbers(n: int = 5) -> Generator

发射带有延迟的数字1..n，展示流媒体功能。

live_data_feed(duration: int = 10) -> Generator

在指定持续时间内流式传输实时系统指标（CPU、内存等）。

progress_tracker(task: str, steps: int = 8) -> Generator

通过状态更新跟踪长时间运行的任务的进度。

测试服务器

使用MCP检查器

# Install and run MCP Inspector
npx @modelcontextprotocol/inspector

# Connect to: http://localhost:3001/sse

使用curl

# Initialize session
curl -X POST http://localhost:3001/mcp \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -d '{
    "jsonrpc": "2.0",
    "id": 1,
    "method": "initialize",
    "params": {
      "protocolVersion": "2024-11-05",
      "capabilities": {},
      "clientInfo": {"name": "test-client", "version": "1.0.0"}
    }
  }'

# List available tools
curl -X POST http://localhost:3001/mcp \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -d '{
    "jsonrpc": "2.0",
    "id": 2,
    "method": "tools/list"
  }'

# Call the add tool
curl -X POST http://localhost:3001/mcp \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -d '{
    "jsonrpc": "2.0",
    "id": 3,
    "method": "tools/call",
    "params": {
      "name": "add",
      "arguments": {"a": 15, "b": 27}
    }
  }'

# Test streaming (use -N for real-time output)
curl -N -X POST http://localhost:3001/mcp \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -d '{
    "jsonrpc": "2.0",
    "id": 4,
    "method": "tools/call",
    "params": {
      "name": "stream_numbers",
      "arguments": {"n": 5}
    }
  }'

配置

环境变量：

• MCP_HOST -服务器主机（默认值：“0.0.0.0”）
• MCP_PORT -服务器端口（默认：3001）

建筑

此服务器使用：

• FastMCP -现代MCP服务器框架
• 流式HTTP传输 -基于HTTP，支持SSE
• 无状态模式 -用于水平缩放
• 生成函数 -用于随时间流式传输数据

服务器通过以下方式整合所有客户端-服务器交互：

• POST /mcp -用于即时JSON响应
• GET /sse -用于服务器发送的事件（流式传输）

发展

项目结构

mcp-uv-base/
├── .devcontainer/          # Dev container configuration
│   ├── devcontainer.json   # VS Code dev container settings
│   └── Dockerfile          # Container image definition
├── main.py                 # FastMCP server implementation
├── pyproject.toml          # Python project configuration
├── uv.lock                 # Locked dependencies
└── README.md               # This file

添加新工具

@mcp.tool()
def your_new_tool(param: str) -> str:
    \"\"\"Description of your tool.\"\"\"
    # Your implementation here
    return f"Result: {param}"

对于流媒体工具，请使用生成器：

@mcp.tool()
def your_streaming_tool(count: int) -> Generator[Dict[str, Any], None, Dict[str, str]]:
    \"\"\"A streaming tool example.\"\"\"
    for i in range(count):
        yield {"step": i, "data": f"Processing {i}"}
        time.sleep(1)
    return {"status": "completed"}

许可证

MIT许可证-您可以将其作为您自己的MCP服务器的起点。