mcp-utils

mcp-utils

一个用于构建模型上下文协议（MCP）服务的 Python 工具包。

目录

• mcp-utils
  • 目录
  • 概述
  • 主要功能
  • 安装
  • 要求
    • 可选依赖项
  • 用法
    • 基本 MCP 服务
    • Flask 示例
    • SQLAlchemy 事务处理示例
    • 使用 Gunicorn 运行
  • 连接 MCP 客户端
    • Cursor
    • Claude Desktop
      • 通过 Smithery 安装
      • 通过 PyPI 安装
  • 贡献
  • 相关项目
  • 许可证
  • 使用 MCP Inspector 测试
    • 安装
    • 用法

概述

mcp-utils 提供用于使用 Python 构建符合 MCP 标准的服务的实用工具和辅助函数，重点支持使用 Flask 的同步实现。该软件包面向希望在现有 Python 应用中实现 MCP 服务、又不想处理异步代码复杂性的开发者。

主要功能

• 用于实现 MCP 服务的基本工具
• 支持服务器发送事件（SSE）
• 用于 MCP 端点的简单装饰器
• 同步实现
• 支持 HTTP 协议
• Redis 响应队列
• 全面的 MCP schema Pydantic 模型
• 内置验证和文档

安装

pip install mcp-utils

要求

• Python 3.10+
• Pydantic 2

可选依赖项

• Flask（用于 Web 服务）
• Gunicorn（用于生产环境部署）
• Redis（用于响应队列）

用法

基本 MCP 服务

以下是创建 MCP 服务的简单示例：

from mcp_utils.core import MCPServer
from mcp_utils.schema import GetPromptResult, Message, TextContent, CallToolResult

# Create a basic MCP server
mcp = MCPServer("example", "1.0")

@mcp.prompt()
def get_weather_prompt(city: str) -> GetPromptResult:
    return GetPromptResult(
        description="Weather prompt",
        messages=[
            Message(
                role="user",
                content=TextContent(
                    text=f"What is the weather like in {city}?",
                ),
            )
        ],
    )

@mcp.tool()
def get_weather(city: str) -> str:
    return "sunny"

Flask 示例

在生产环境中，你可以使用简单的 Flask 应用搭配 MCP 服务，并从 2025-06-18 版本开始支持 Streamable HTTP。

from flask import Flask, Response, url_for, request

# Create Flask app and MCP server with Redis queue
app = Flask(__name__)
mcp = MCPServer(
    "example",
    "1.0",
)

@app.route("/mcp", methods=["POST"])
def mcp_route():
    response = mcp.handle_message(request.get_json())
    return jsonify(response.model_dump(exclude_none=True))


if __name__ == "__main__":
    app.run(debug=True)

SQLAlchemy 事务处理示例

在生产环境中，你可以将 MCP 服务与 Flask、Redis 和 SQLAlchemy 集成，以改进消息处理和数据库事务管理：

from flask import Flask, request
from sqlalchemy.orm import Session
from sqlalchemy import create_engine

# Create engine for PostgreSQL database
engine = create_engine("postgresql://user:pass@localhost/dbname")

# Create Flask app and MCP server with Redis queue
app = Flask(__name__)
mcp = MCPServer(
    "example",
    "1.0",
)

@app.route("/mcp", methods=["POST"])
def mcp_route():
    with Session(engine) as session:
        try:
            response = mcp.handle_message(request.get_json())
            session.commit()
        except:
            session.rollback()
            raise
        else:
            return jsonify(response.model_dump(exclude_none=True))


if __name__ == "__main__":
    app.run(debug=True)

如需包含日志设置和会话管理的更完整示例，请查看仓库中的示例 Flask 应用。

使用 Gunicorn 运行

即使在本地运行，Gunicorn 也是更好的选择。要使用 gunicorn 运行应用：

from gunicorn.app.base import BaseApplication

class FlaskApplication(BaseApplication):
    def __init__(self, app, options=None):
        self.options = options or {}
        self.application = app
        super().__init__()

    def load_config(self):
        config = {
            key: value
            for key, value in self.options.items()
            if key in self.cfg.settings
        }
        for key, value in config.items():
            self.cfg.set(key.lower(), value)

    def load(self):
        return self.application


if __name__ == "__main__":
    handler = logging.StreamHandler(sys.stdout)
    formatter = logging.Formatter("[%(asctime)s] [%(levelname)s] %(name)s: %(message)s")
    handler.setFormatter(formatter)
    logger.addHandler(handler)
    options = {
        "bind": "0.0.0.0:9000",
        "workers": 1,
        "worker_class": "gevent",
        "loglevel": "debug",
    }
    FlaskApplication(app, options).run()

连接 MCP 客户端

Cursor

• 编辑 MCP 设置并将以下内容添加到配置中

{
  "mcpServers": {
    "server-name": {
      "url": "http://localhost:9000/mcp"
    }
  }
}

Claude Desktop

截至本文撰写时，Claude Desktop 不支持通过 SSE 使用 MCP，仅支持 stdio。要将 Claude Desktop 连接到 MCP 服务，你需要使用 mcp-proxy。

Claude Desktop 配置示例：

{
  "mcpServers": {
    "weather": {
      "command": "/Users/yourname/.local/bin/mcp-proxy",
      "args": ["http://127.0.0.1:9000/sse"]
    }
  }
}

通过 Smithery 安装

要通过 Smithery 自动为 Claude Desktop 安装 MCP Proxy：

npx -y @smithery/cli install mcp-proxy --client claude

通过 PyPI 安装

该软件包的稳定版本可在 PyPI 仓库中获取。你可以使用以下命令安装：

# Option 1: With uv (recommended)
uv tool install mcp-proxy

# Option 2: With pipx (alternative)
pipx install mcp-proxy

安装后，你可以使用 mcp-proxy 命令运行服务。

贡献

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

相关项目

• MCP Python SDK - 用于 MCP 的官方异步 Python SDK
• mcp-proxy - 用于连接 Claude Desktop 与 MCP 服务的代理工具

许可证

MIT 许可证

使用 MCP Inspector 进行测试

MCP Inspector 是一款用于测试和调试 MCP 服务的实用工具。它提供了一个 Web 界面，用于检查和测试 MCP 服务端点。

安装

使用 npm 安装 MCP Inspector：

npm install -g @modelcontextprotocol/inspector

用法

1. 启动 MCP 服务（例如上文的 Flask 示例）
2. 运行 MCP Inspector：

git clone [email protected]:modelcontextprotocol/inspector.git
cd inspector
npm run build
npm start

3. 打开浏览器并访问 http://127.0.0.1:6274/
4. 输入 MCP 服务 URL（例如 http://localhost:9000/sse）
5. 使用检查器执行以下操作：
   • 将传输类型更改为 SSE
   • 测试服务连接
   • 监控 SSE 事件
   • 发送测试消息
   • 调试响应

此工具在开发过程中尤其有用，可确保 MCP 服务实现正常运行并符合协议规范。