mcp-utils

一个用于构建模型上下文协议 (MCP) 服务器的 Python 实用工具包。

目录

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

概述

mcp-utils 提供了用于在 Python 中构建符合 MCP 规范的服务器的实用工具和助手，重点是使用 Flask 的同步实现。 此软件包专为希望在其现有 Python 应用程序中实现 MCP 服务器的开发人员而设计，无需异步代码的复杂性。

主要特性

• 用于 MCP 服务器实现的基本实用工具
• 服务器发送事件 (SSE) 支持
• 用于 MCP 端点的简单装饰器
• 同步实现
• HTTP 协议支持
• Redis 响应队列
• 用于 MCP 模式的全面 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

# 创建一个基本的 MCP 服务器
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 与 Redis 示例

对于生产用途，您可以将 MCP 服务器与 Flask 和 Redis 集成，以获得更好的消息处理：

from flask import Flask, Response, url_for, request
import redis
from mcp_utils.queue import RedisResponseQueue

# 设置 Redis 客户端
redis_client = redis.Redis(host="localhost", port=6379, db=0)

# 使用 Redis 队列创建 Flask 应用程序和 MCP 服务器
app = Flask(__name__)
mcp = MCPServer(
    "example",
    "1.0",
    response_queue=RedisResponseQueue(redis_client)
)

@app.route("/sse")
def sse():
    session_id = mcp.generate_session_id()
    messages_endpoint = url_for("message", session_id=session_id)
    return Response(
        mcp.sse_stream(session_id, messages_endpoint),
        mimetype="text/event-stream"
    )


@app.route("/message/<session_id>", methods=["POST"])
def message(session_id):
    mcp.handle_message(session_id, request.get_json())
    return "", 202


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
import redis
from mcp_utils.queue import RedisResponseQueue

# 设置 Redis 客户端
redis_client = redis.Redis(host="localhost", port=6379, db=0)

# 为 PostgreSQL 数据库创建引擎
engine = create_engine("postgresql://user:pass@localhost/dbname")

# 使用 Redis 队列创建 Flask 应用程序和 MCP 服务器
app = Flask(__name__)
mcp = MCPServer(
    "example",
    "1.0",
    response_queue=RedisResponseQueue(redis_client)
)

@app.route("/message/<session_id>", methods=["POST"])
def message(session_id):
    with Session(engine) as session:
        try:
            mcp.handle_message(session_id, request.get_json())
            session.commit()
            return "", 202
        except Exception as e:
            session.rollback()
            raise

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

有关包括日志设置和会话管理的更全面的示例，请查看存储库中的 示例 Flask 应用程序。

连接 MCP 客户端

Claude Desktop

目前，只有 Claude Desktop（不是 claude.ai）可以连接到 MCP 服务器。 截至撰写本文时，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 存储库中找到。 您可以使用以下命令安装它：

# 选项 1：使用 uv（推荐）
uv tool install mcp-proxy

# 选项 2：使用 pipx（替代）
pipx install mcp-proxy

安装完成后，您可以使用 mcp-proxy 命令运行服务器。

贡献

欢迎贡献！ 请随时提交拉取请求。

相关项目

• 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 git@github.com: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 服务器实现 工作正常并符合协议规范。