mcp-utils

mcp-utils

Model Context Protocol（MCP）サーバーを構築するための Python ユーティリティパッケージです。

目次

• mcp-utils
  • 目次
  • 概要
  • 主な機能
  • インストール
  • 要件
    • オプションの依存関係
  • 使い方
    • 基本的な MCP サーバー
    • Flask の例
    • SQLAlchemy のトランザクション処理の例
    • Gunicorn での実行
  • MCP クライアントとの接続
    • Cursor
    • Claude Desktop
      • Smithery 経由でのインストール
      • PyPI 経由でのインストール
  • コントリビューション
  • 関連プロジェクト
  • ライセンス
  • MCP Inspector によるテスト
    • インストール
    • 使い方

概要

mcp-utils は、Python で MCP に準拠したサーバーを構築するためのユーティリティとヘルパーを提供します。特に Flask を使用した同期実装に重点を置いています。このパッケージは、非同期コードの複雑さを避け、既存の Python アプリケーションに MCP サーバーを実装したい開発者向けに設計されています。

主な機能

• MCP サーバー実装の基本的なユーティリティ
• Server-Sent Events（SSE）のサポート
• MCP エンドポイント向けのシンプルなデコレーター
• 同期実装
• HTTP プロトコルのサポート
• Redis 応答キュー
• MCP スキーマ向けの包括的な Pydantic モデル
• 組み込みの検証とドキュメント

インストール

pip install mcp-utils

要件

• Python 3.10+
• Pydantic 2

オプションの依存関係

• Flask（ウェブサーバー用）
• Gunicorn（本番環境へのデプロイ用）
• Redis（応答キュー用）

使い方

基本的な MCP サーバー

MCP サーバーを作成する簡単な例を紹介します。

from mcp_utils.core import MCPServer
from mcp_utils.schema import GetPromptResult, Message, TextContent, CallToolResult

# Create a basic MCP server
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 の例

本番環境では、シンプルな Flask アプリで MCP サーバーを使用して、 バージョン 2025-06-18 からの Streamable HTTP をサポートできます。

from flask import Flask, Response, url_for, request

# Create Flask app and MCP server with Redis queue
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)

SQLAlchemy のトランザクション処理の例

本番環境では、MCP サーバーを Flask、Redis、SQLAlchemy と統合することで、メッセージ処理とデータベースのトランザクション管理を改善できます。

from flask import Flask, request
from sqlalchemy.orm import Session
from sqlalchemy import create_engine

# Create engine for PostgreSQL database
engine = create_engine("postgresql://user:pass@localhost/dbname")

# Create Flask app and MCP server with Redis queue
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 を使用する方法がおすすめです。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()

MCP クライアントとの接続

Cursor

• MCP の設定を編集し、構成に追加します。

{
  "mcpServers": {
    "server-name": {
      "url": "http://localhost:9000/mcp"
    }
  }
}

Claude Desktop

この記事の執筆時点では、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 リポジトリで入手できます。次のコマンドでインストールできます。

# Option 1: With uv (recommended)
uv tool install mcp-proxy

# Option 2: With pipx (alternative)
pipx install mcp-proxy

インストール後は、mcp-proxy コマンドを使用してサーバーを実行できます。

コントリビューション

コントリビューションを歓迎します。お気軽にプルリクエストを送信してください。

関連プロジェクト

• MCP Python SDK - MCP の公式非同期 Python SDK
• mcp-proxy - Claude Desktop を MCP サーバーに接続するためのプロキシツール

ライセンス

MIT License

MCP Inspector を使ったテスト

MCP Inspector は、MCP サーバーのテストとデバッグに役立つツールです。MCP サーバーのエンドポイントを調査、テストするためのウェブインターフェイスを提供します。

インストール

npm を使って MCP Inspector をインストールします。

npm install -g @modelcontextprotocol/inspector

使用方法

1. MCP サーバーを起動します（例：上記の Flask の例）。
2. MCP Inspector を実行します。

git clone [email protected]: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. Inspector を使って、以下を行います。
   • トランスポートタイプを SSE に変更する
   • サーバー接続をテストする
   • SSE イベントを監視する
   • テストメッセージを送信する
   • 応答をデバッグする

このツールは、開発中に MCP サーバーの実装が正しく動作し、プロトコル仕様に準拠していることを確認するのに特に役立ちます。