一个用于构建模型上下文协议(MCP)服务器的Python工具包。
mcp-utils 提供了在Python中构建符合MCP规范的服务器所需的实用工具和辅助功能,重点是使用Flask实现同步操作。此包旨在帮助开发者在其现有的Python应用程序中实现MCP服务器,而无需处理异步代码的复杂性。
pip install mcp-utils
这是一个创建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="天气提示",
messages=[
Message(
role="user",
content=TextContent(
text=f"{city}的天气如何?",
),
)
],
)
@mcp.tool()
def get_weather(city: str) -> str:
return "晴朗"
对于生产用途,你可以使用简单的Flask应用与MCP服务器,并支持从版本2025-06-18开始的流式HTTP。
from flask import Flask, Response, url_for, request
# 创建Flask应用和带有Redis队列的MCP服务器
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)
为了更好地处理消息和数据库事务管理,你可以将MCP服务器与Flask、Redis和SQLAlchemy集成:
from flask import Flask, request
from sqlalchemy.orm import Session
from sqlalchemy import create_engine
# 为PostgreSQL数据库创建引擎
engine = create_engine("postgresql://user:pass@localhost/dbname")
# 创建Flask应用和带有Redis队列的MCP服务器
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运行应用:
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()
{
"mcpServers": {
"server-name": {
"url": "http://localhost:9000/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自动安装MCP Proxy:
npx -y @smithery/cli install mcp-proxy --client claude
该包的稳定版本可以在PyPI存储库中找到。你可以使用以下命令进行安装:
# 选项1:使用uv(推荐)
uv tool install mcp-proxy
# 选项2:使用pipx(替代方案)
pipx install mcp-proxy
安装后,你可以使用mcp-proxy命令运行服务器。
欢迎贡献!请随时提交Pull Request。
MIT许可证
MCP Inspector 是一个有用的工具,用于测试和调试MCP服务器。它提供了一个Web界面来检查和测试MCP服务器端点。
使用npm安装MCP Inspector:
npm install -g @modelcontextprotocol/inspector
git clone git@github.com:modelcontextprotocol/inspector.git
cd inspector
npm run build
npm start
http://127.0.0.1:6274/http://localhost:9000/sse)这个工具在开发过程中特别有用,确保你的MCP服务器实现正确且符合协议规范。