ezmcp

🚀 ezmcp

ezmcp是一个易于使用的、专为SSE设计的MCP服务器框架。它能让创建兼容MCP的服务器变得简单，并且采用了FastAPI式的语法。

🚀 快速开始

from ezmcp import ezmcp, TextContent

# 创建一个ezmcp应用程序
app = ezmcp("my-app")

# 定义一个工具
@app.tool(description="Echo a message back to the user")
async def echo(message: str):
    """回显用户的消息."""
    return [TextContent(type="text", text=f"Echo: {message}")]

# 运行应用程序
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

一旦服务器运行，你可以：

• 访问交互式文档页面 http://localhost:8000/docs
• 连接到SSE端点 http://localhost:8000/sse

✨ 主要特性

• FastAPI风格的装饰器API用于定义MCP工具
• 自动参数验证和类型转换
• 从函数签名自动生成工具模式
• 内置对SSE传输的支持
• FastAPI风格的中间件支持
• 与现有Starlette应用程序轻松集成
• 交互式文档页面用于探索和测试工具

📦 安装指南

pip install ezmcp

💻 使用示例

基础用法

from ezmcp import ezmcp, TextContent

# 创建一个ezmcp应用程序
app = ezmcp("my-app")

# 定义一个工具
@app.tool(description="Echo a message back to the user")
async def echo(message: str):
    """回显用户的消息."""
    return [TextContent(type="text", text=f"Echo: {message}")]

# 运行应用程序
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

高级用法

from starlette.requests import Request

from ezmcp import TextContent, ezmcp

app = ezmcp("my-app")

@app.middleware
async def process_time_middleware(request: Request, call_next):
    """在响应中添加处理时间头."""
    import time
    start_time = time.perf_counter()
    response = await call_next(request)
    process_time = time.perf_counter() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

@app.tool(description="Echo a message back to the user")
async def echo(message: str):
    """回显用户的消息."""
    return [TextContent(type="text", text=f"Echo: {message}")]

有关中间件的更多信息，请参阅middleware.md。

📚 详细文档

更多详细信息请查阅ezmcp/README.md文件。

📄 许可证

MIT

🔧 其他命令

• 使用测试环境安装 pdm install -G test
• 使用开发环境安装 pdm install -G dev