MCP服务器框架

ezmcp是一个轻量级的框架,用于使用类似FastAPI的语法创建MCP兼容的服务器,支持SSE传输和工具定义。

  • by jujumilk3
  • Python
  • 本地部署
  • 23 星标
  • 更新于
在 GitHub 查看 jujumilk3/ezmcp

使用说明

ezmcp

易于使用的MCP服务器框架,专为SSE设计。

概述

ezmcp是一个轻量级的框架,可以使用类似FastAPI的语法轻松创建与MCP兼容的服务器。它提供了一个简单的基于装饰器的API,用于定义MCP客户端可以调用的工具。

特性

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

安装

pip install ezmcp

快速开始

from ezmcp import ezmcp, TextContent

# Create an ezmcp application
app = ezmcp("my-app")

# Define a tool
@app.tool(description="Echo a message back to the user")
async def echo(message: str):
    """Echo a message back to the user."""
    return [TextContent(type="text", text=f"Echo: {message}")]

# Run the application
if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

服务器运行后,您可以:

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

中间件

ezmcp支持类似于FastAPI的中间件,允许您添加在整个应用程序中应用的行为。

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):
    """Add a header with the processing time."""
    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):
    """Echo a message back to the user."""
    return [TextContent(type="text", text=f"Echo: {message}")]

有关中间件的更多信息,请参阅 中间件文档.

文档

有关更详细的文档,请参阅 ezmcp/README.md 文件。

许可证

麻省理工学院

命令

  • 安装测试 pdm install -G test
  • 为dev安装 pdm install -G dev