ezmcp

ezmcp

SSE に特化した、使いやすい MCP サーバーフレームワークです。

概要

ezmcp は、FastAPI のような構文を使用して、MCP 互換サーバーを簡単に作成できる軽量フレームワークです。MCP クライアントから呼び出せるツールを定義するための、シンプルなデコレーターベースの API を提供します。

特長

• MCP ツールを定義するための FastAPI スタイルのデコレーター API
• パラメーターの自動検証と型変換
• 関数のシグネチャからツールスキーマを自動生成
• 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 でインタラクティブなドキュメントにアクセスする
• http://localhost:8000/sse の 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 ファイルを参照してください。

ライセンス

MIT

コマンド

• テスト用にインストール pdm install -G test
• 開発用にインストール pdm install -G dev