MCP Streamable HTTP Demo

English | 日本語

stdio向けのタスクToolを、複数クライアントがURLで利用できるStatefulな Streamable HTTP MCPサーバーとして実行するサンプルです。Toolと業務ロジックを transportから分離し、セッション、SSE、Host・Origin検証、Dockerのネットワーク境界を 実際に確認できます。

このリポジトリは記事 「MCPサーバーをStreamable HTTPでリモート化する方法」 の完成形コードです。

5分で試す

前提は uv だけです。uvがPython 3.13も用意します。

git clone https://github.com/yunosuke-github/mcp-streamable-http-demo.git
cd mcp-streamable-http-demo
cp .env.example .env
uv sync
uv run --env-file .env task-mcp-server

サーバーは http://127.0.0.1:8000/mcp で待機します。別のターミナルで、 2つのMCPクライアントを同時に実行してください。

uv run python clients/concurrent_clients.py

次の4点が表示されれば成功です。

• AとBで異なるセッションIDの短縮値

• Aが作成したタスク

• Bが取得したタスク一覧

• Bの一覧にAのタスクが含まれていること

クライアントは15秒でタイムアウトし、Toolエラーや想定外の構造化結果を成功扱いしません。

Related MCP server: Levitate

Dockerで試す

Docker内部ではサーバーを 0.0.0.0:8000 へバインドしますが、ホスト側で公開するのは 127.0.0.1:8000だけです。

docker compose up --build -d
uv run python clients/concurrent_clients.py
docker compose down

8000番が使用中なら、ホスト側の公開ポートだけを変更できます。

MCP_PUBLISH_PORT=8766 docker compose up --build -d
MCP_URL=http://127.0.0.1:8766/mcp uv run python clients/concurrent_clients.py
docker compose down

Composeの公開アドレスは次のコマンドでも確認できます。

docker compose config

コンテナはUID 10001の非rootユーザーで動きます。

docker compose exec task-mcp id -u
# 10001

公開するTool

RESTの /tasks エンドポイントは作りません。/mcp内を流れるMCPのJSON-RPCメッセージを 公式Python SDKが処理し、登録済みToolへ委譲します。

アーキテクチャ

flowchart LR
    A[Client A] -->|Streamable HTTP| M[/mcp/]
    B[Client B] -->|Streamable HTTP| M
    M --> S[FastMCP stateful sessions]
    S --> T[Tool adapters]
    T --> D[Shared TaskService]
    D --> L[asyncio.Lock]

TaskServiceはHTTP、セッション、Dockerを知りません。tools/tasks.pyが薄いアダプターに なり、server.pyだけがFastMCPとtransport設定を担当します。同じプロセス内では asyncio.Lockが同時更新を守りますが、複数ワーカーや複数コンテナでは共有データベースが必要です。

MCPセッションとタスクデータは別物です。AとBは異なるセッションIDを持ちますが、同じ TaskServiceを利用するためタスクを共有します。セッションを終了してもタスクは削除されません。

セキュリティ境界

• TransportSecuritySettingsがHostとOriginを許可リストで検証します。

• 不正なHostはHTTP 421、不正なOriginはHTTP 403になります。

• POSTのContent-TypeがJSONでなければHTTP 400になります。

• Composeはホストの127.0.0.1だけへポートを公開します。

• コンテナは非rootで動作します。

• セッションID、CORS、Origin検証、TLS、Dockerはいずれも認証の代わりにはなりません。

ブラウザから直接接続する場合は、別途CORS middlewareで必要なOrigin、GET・POST・DELETE、 リクエストヘッダー、公開するMcp-Session-Idレスポンスヘッダーを限定してください。 このサンプルはブラウザ接続を実装していません。

構成

mcp-streamable-http-demo/
├── clients/concurrent_clients.py  # 2つのStatefulクライアント
├── src/task_mcp/
│   ├── server.py                  # FastMCPとStreamable HTTP設定
│   ├── settings.py                # 環境変数と許可リスト
│   ├── services/task_service.py   # transport非依存の業務ロジック
│   └── tools/tasks.py             # MCP Toolアダプター
├── tests/                         # Unit・実HTTP・セキュリティ試験
├── Dockerfile                     # Python 3.13、uv、非root実行
├── compose.yaml                   # loopback限定のポート公開
└── pyproject.toml                 # MCP SDK v1と開発Tool

設定

.envを読み込むときは、サーバー起動コマンドへ--env-file .envを付けます。Composeは compose.yamlのenvironmentを使用します。秘密情報を.env.exampleへ入れないでください。

MCP Inspector

サーバー起動後、公式InspectorでtransportにStreamable HTTP、URLに http://127.0.0.1:8000/mcpを指定します。

npx -y @modelcontextprotocol/inspector

開発とテスト

uv run ruff format --check .
uv run ruff check .
uv run mypy
uv run pytest --cov=task_mcp --cov-report=term-missing
uv build

テストはTaskServiceの同時作成、3つのTool、異なるStatefulセッション間のデータ共有、 421/403/400のHTTP境界、Toolエラー、非機密なサーバー情報を確認します。

トラブルシューティング

• Connection refused: サーバー起動ターミナルを確認し、MCP_URLとポートを合わせます。

• Invalid Host header: 接続URLのHostをMCP_ALLOWED_HOSTSへ完全一致または:*形式で追加します。

• Invalid Origin header: 必要なOriginだけをMCP_ALLOWED_ORIGINSへ追加します。

• Dockerで到達できない: コンテナ内のMCP_HOSTが0.0.0.0か確認します。

• ポート8000が使用中: ローカル実行では.envのMCP_PORTを変えます。Composeでは MCP_PUBLISH_PORT=8766を付け、クライアントのMCP_URLも8766へ合わせます。

本番利用について

このサンプルをそのまま本番へ出さないでください。リモート公開にはOAuth 2.1ベースの認証・認可、 HTTPS、永続ストレージ、監視、レート制限、秘密情報管理、プロキシのSSE設定、切断・再送・ キャンセル・graceful shutdownの設計が必要です。