MCPサーバーをStreamable HTTPでリモート化する方法：Python・SSE・Docker対応

セキュリティ上の注意
本記事では認証を実装しません。構築したMCPサーバーは、localhostまたは外部からアクセスできない閉じた検証環境でのみ使用してください。パブリックIPやインターネットへ公開してはいけません。

stdioからHTTPサービスへ移行する

stdio形式のMCPサーバーは、ローカルツールとして非常に扱いやすい構成です。MCPクライアントがサーバーを子プロセスとして起動し、標準入力と標準出力を使って1対1でJSON-RPCメッセージを交換します。ポートを開く必要がなく、サーバープロセスの寿命もクライアントが管理できます。

しかし、別の端末、Webアプリ、複数のAIエージェントから同じサーバーを使いたくなると、stdioのプロセスモデルが制約になります。Streamable HTTPへ移行すると、MCPサーバーは独立して待ち受けるネットワークサービスになり、クライアントはコマンドではなくURLへ接続します。

重要なのは、変更が次の一行で完結しないことです。

mcp.run(transport="streamable-http")

この変更によって、プロセス管理、同時接続、セッション、切断、タイムアウト、Origin、TLS、認証といったWebサービスの責任が加わります。本記事ではタスク管理サーバーを題材に、http://127.0.0.1:8000/mcpへ複数のクライアントを接続し、Dockerで再現可能にするところまで実装します。

実装は、2026年7月16日時点で安定版と案内されている公式Python SDKのv1系を対象にします。mainブランチで説明されているv2はプレリリースで、破壊的変更があり得るため混在させません。依存関係にはmcp>=1.27,<2という上限を設定し、実際の解決バージョンはuv.lockで固定します。MCP Python SDK

今回作る構成

ホストではPythonクライアントA、クライアントB、必要に応じてMCP Inspectorを実行します。MCPサーバーは単一のDockerコンテナで動かし、ホストの127.0.0.1:8000だけへポートを公開します。

Streamable HTTPサーバーは単一のMCPエンドポイントでPOSTとGETを処理します。Stateful構成でクライアントによるセッション終了をサポートする場合はDELETEも扱います。MCP transport specification

今回行うことは次のとおりです。

• 既存のToolと業務ロジックを維持する
• 起動トランスポートをStreamable HTTPへ変更する
• Statefulなセッションを観察する
• JSONレスポンスとSSEレスポンスの用途を整理する
• Docker内部とホスト側の待受境界を分ける
• 2つのクライアントを同時に接続する
• HostとOriginを許可リストで検証する

OAuth 2.1、TLS終端、ロードバランサー、Redis、複数コンテナへの水平分散、監視、レート制限は対象外です。Docker化は本番対応の完了を意味しません。

stdioとStreamable HTTPの違い

stdioでは、クライアントがMCPサーバーを子プロセスとして起動します。サーバーはstdinから改行区切りのJSON-RPCメッセージを読み、stdoutへMCPメッセージを返します。ログはstderrへ出力しなければなりません。stdoutへ通常のログを混ぜると、プロトコルメッセージとして解釈されて通信が壊れます。

Streamable HTTPでは、サーバーはクライアントから独立したプロセスです。クライアントが終了してもサーバーは待ち受けを続け、別のクライアントが同じURLへ接続できます。

Streamable HTTP化はREST API化でもありません。POST /tasksやGET /tasksを実装するのではなく、基本的には/mcpという一つのエンドポイント内をJSON-RPCメッセージが流れます。tools/listやtools/callをMCP SDKが処理し、その先で登録済みのToolを呼び出します。

完成コード：この記事の実装、セットアップ、テスト一式はGitHubリポジトリで公開しています。以降のコードブロックに表示するファイル名は、このリポジトリ内のパスに対応します。

Toolとトランスポートを分離する

HTTP対応のためにToolへHTTPヘッダーやSSE処理を書き始めると、業務ロジックが特定のトランスポートへ結合します。タスク保存処理はサービス層へ置き、Toolは入力をサービスへ渡す薄いアダプターにします。

mcp-streamable-http-demo/
├── pyproject.toml
├── uv.lock
├── Dockerfile
├── compose.yaml
├── .dockerignore
├── src/task_mcp/
│   ├── server.py
│   ├── settings.py
│   ├── services/task_service.py
│   └── tools/tasks.py
└── clients/concurrent_clients.py

これはファイル構成の提示であり、処理フローの図ではないためMermaidにはしません。

まず、複数の非同期Tool呼び出しから安全に利用できるインメモリサービスを作ります。

import asyncio
from dataclasses import asdict, dataclass
from uuid import uuid4


@dataclass(frozen=True)
class Task:
    id: str
    title: str


class TaskService:
    def __init__(self) -> None:
        self._tasks: list[Task] = []
        self._lock = asyncio.Lock()

    async def create(self, title: str) -> dict[str, str]:
        normalized = title.strip()
        if not normalized:
            raise ValueError("title must not be empty")

        task = Task(id=str(uuid4()), title=normalized)
        async with self._lock:
            self._tasks.append(task)
        return asdict(task)

    async def list_all(self) -> list[dict[str, str]]:
        async with self._lock:
            return [asdict(task) for task in self._tasks]

asyncio.Lockが守るのは同じプロセス内の同時更新だけです。ワーカーやコンテナを増やす場合、各プロセスは別のリストと別のロックを持つため、データベースなどの共有ストアが必要です。

Toolはサービスへ委譲します。

from task_mcp.services.task_service import TaskService

service = TaskService()


async def create_task(title: str) -> dict[str, str]:
    """Create a task in the shared in-memory store."""
    return await service.create(title)


async def list_tasks() -> list[dict[str, str]]:
    """List tasks shared by all client sessions."""
    return await service.list_all()

ここにはHTTPリクエスト、セッションヘッダー、CORS、Docker固有処理がありません。同じ関数をstdioサーバーや単体テストからも利用できます。

Streamable HTTPサーバーを実装する

ホスト、ポート、許可するHostとOriginを環境変数から読みます。Originの許可値は完全一致を基本とし、検証環境で必要なものだけを指定します。

import os
from dataclasses import dataclass, field


def csv_env(name: str, default: str) -> list[str]:
    return [value.strip() for value in os.getenv(name, default).split(",") if value.strip()]


def port_env(name: str = "MCP_PORT", default: str = "8000") -> int:
    value = int(os.getenv(name, default))
    if not 1 <= value <= 65535:
        raise ValueError(f"{name} must be between 1 and 65535")
    return value


@dataclass(frozen=True)
class Settings:
    host: str = field(default_factory=lambda: os.getenv("MCP_HOST", "127.0.0.1"))
    port: int = field(default_factory=port_env)
    allowed_hosts: list[str] = field(
        default_factory=lambda: csv_env("MCP_ALLOWED_HOSTS", "127.0.0.1:*,localhost:*,[::1]:*")
    )
    allowed_origins: list[str] = field(
        default_factory=lambda: csv_env(
            "MCP_ALLOWED_ORIGINS",
            "http://127.0.0.1:*,http://localhost:*,http://[::1]:*",
        )
    )

次にFastMCPへToolを登録します。本記事ではセッションIDとSSEを観察するため、stateless_http=False、json_response=FalseのStateful構成を明示します。

from contextlib import suppress

from mcp.server.fastmcp import FastMCP
from mcp.server.transport_security import TransportSecuritySettings

from task_mcp.settings import Settings
from task_mcp.tools.tasks import create_task, list_tasks


def create_server(settings: Settings | None = None) -> FastMCP:
    configured = settings or Settings()
    server = FastMCP(
        "Task Server",
        host=configured.host,
        port=configured.port,
        streamable_http_path="/mcp",
        stateless_http=False,
        json_response=False,
        transport_security=TransportSecuritySettings(
            enable_dns_rebinding_protection=True,
            allowed_hosts=configured.allowed_hosts,
            allowed_origins=configured.allowed_origins,
        ),
    )
    server.add_tool(create_task)
    server.add_tool(list_tasks)

    @server.tool()
    def get_server_info() -> dict[str, object]:
        """Return non-sensitive runtime configuration."""

        return {
            "transport": "streamable-http",
            "host": configured.host,
            "port": configured.port,
            "stateful": True,
            "storage": "process-memory",
        }

    return server


settings = Settings()
mcp = create_server(settings)


def main() -> None:
    with suppress(KeyboardInterrupt):
        mcp.run(transport="streamable-http")


if __name__ == "__main__":
    main()

公式SDKのv1系はTransportSecuritySettingsでHostとOriginを検証できます。不正なHostには421、不正なOriginには403を返す実装です。また、最近のv1系ではlocalhostへバインドしたFastMCPに保護設定を自動適用しますが、コンテナでは0.0.0.0へバインドするため、ここでは設定を明示しています。Python SDK transport security source

ローカルで起動する場合は次の設定になります。

MCP_HOST=127.0.0.1
MCP_PORT=8000
MCP_PUBLISH_PORT=8000
MCP_ALLOWED_HOSTS=127.0.0.1:*,localhost:*,[::1]:*
MCP_ALLOWED_ORIGINS=http://127.0.0.1:*,http://localhost:*,http://[::1]:*

curlはHTTPレベルの疎通確認には使えますが、完全なMCPクライアントの代わりにはなりません。初期化、プロトコルバージョンの合意、JSON-RPC ID、セッションID、通知を正しく扱う必要があるからです。

MCPセッションを理解する

MCPセッションは、クライアントとサーバー間で論理的に関連する通信のまとまりです。StatefulサーバーはInitializeResultを返すHTTPレスポンスにMCP-Session-Idを付けられます。クライアントは以降のHTTPリクエストに同じヘッダーを含めます。

sequenceDiagram
    accTitle: Streamable HTTPセッションの確立と終了
    accDescr: クライアントは初期化要求を送り、サーバーからセッションIDを受け取る。その後のTool一覧取得とTool実行には同じIDを付け、不要になった時点でDELETEにより終了を要求する。
    participant C as MCPクライアント
    participant S as MCPサーバー
    C->>S: POST /mcp InitializeRequest
    S-->>C: InitializeResultとMCP-Session-Id
    C->>S: POST tools/listとセッションID
    S-->>C: Tool定義
    C->>S: POST tools/callとセッションID
    S-->>C: Tool結果
    C->>S: DELETE /mcpとセッションID
    S-->>C: セッション終了

セッション確立と終了

仕様上、セッションIDを要求するサーバーは、初期化後のリクエストにIDがなければ400を返すことが推奨されています。終了済みまたは無効なIDには404を返し、404を受けたクライアントは新しいInitializeRequestからセッションを作り直します。DELETEをサポートしないサーバーは405を返せます。MCP transport specification

ただし、セッションIDは認証情報ではありません。IDが存在しても、接続者の本人性、Toolの実行権限、データの閲覧権限は証明されません。セッションは通信状態を識別し、認証は「誰か」を確認し、認可は「何を許すか」を判断します。

また、MCPセッションとアプリケーションデータも別物です。本実装ではクライアントAとBが別々のセッションIDを持ちますが、両方のToolは同じTaskServiceインスタンスへアクセスします。そのためAが作成したタスクをBが一覧できます。逆に、セッションを削除してもタスク一覧は消えません。

公式SDKは次の構成を選択できます。

Python SDKはスケーラビリティを重視する場合、Stateless + JSONを推奨しています。Building Servers

Stateful構成を複数コンテナへ広げると、最初のリクエストを処理したコンテナと次のリクエストを処理するコンテナが異なる可能性があります。スティッキーセッション、外部セッションストア、またはStateless設計を検討しなければなりません。業務データについても、プロセスメモリではなくデータベースなどへ移す必要があります。

Streamable HTTPにおけるSSE

Streamable HTTPと旧HTTP+SSE transportは同じものではありません。Streamable HTTPは2024-11-05版の旧transportを置き換えましたが、現在のtransportもレスポンス形式としてSSEを使用できます。

クライアントはJSON-RPCメッセージごとに新しいPOSTを送ります。JSON-RPCリクエストを受けたサーバーは、application/jsonで一つのJSONオブジェクトを返すか、text/event-streamでSSEストリームを開始できます。クライアントは両方を受け取れる必要があります。また、クライアントはGETでSSEストリームを開き、サーバーからの通知や要求を待ち受けられます。

JSONレスポンスが適するのは、Toolが短時間で完了し、途中通知が不要で、単純なリクエスト・レスポンスとして閉じる場合です。SSEは、進捗通知、複数メッセージ、サーバーからクライアントへの通知、長時間処理が必要な場合に適します。

SSEを選ぶと、接続タイムアウト、プロキシのバッファリング、切断、再接続、重複配送、キャンセル、コンテナ終了時のドレインも設計対象になります。切断はキャンセルを意味しません。クライアントが処理を取り消す場合は、MCPのキャンセル通知を明示的に送る必要があります。

SSEイベントにIDを付け、再接続時にLast-Event-IDを送れば、サーバーは失われた可能性のあるイベントを再配送できます。ただし、イベントストアや重複に耐える処理はアプリケーション側でも検討が必要です。

Dockerで実行環境を統一する

依存関係は次のように定義します。<2はプレリリース中のv2を誤って取り込まないための境界で、uv.lockをリポジトリへコミットして実際のバージョンを固定します。

[project]
name = "mcp-streamable-http-demo"
version = "0.1.0"
requires-python = ">=3.13"
dependencies = [
  "mcp[cli]>=1.27,<2",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.hatch.build.targets.wheel]
packages = ["src/task_mcp"]

Dockerfileでは依存関係ファイルを先にコピーしてレイヤーキャッシュを利用し、非rootユーザーで実行します。CMDはシグナルがPythonプロセスへ届きやすいexec形式にします。

FROM ghcr.io/astral-sh/uv:python3.13-bookworm-slim

WORKDIR /app

ENV UV_COMPILE_BYTECODE=1 \
    UV_LINK_MODE=copy

COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev --no-install-project

COPY src ./src
RUN uv sync --frozen --no-dev

RUN useradd --create-home --uid 10001 appuser \
    && chown -R appuser:appuser /app
USER appuser

ENV PATH="/app/.venv/bin:$PATH"
EXPOSE 8000

CMD ["python", "-m", "task_mcp.server"]

.git
.venv
__pycache__
.pytest_cache
.env
clients
tests

コンテナ内部で127.0.0.1へバインドすると、待受ソケットはコンテナ自身のループバックに限定され、Dockerのポート転送から到達できません。そのためコンテナ内部では0.0.0.0:8000へバインドします。一方、Composeの公開先はホストの127.0.0.1へ限定します。

services:
  task-mcp:
    build: .
    environment:
      MCP_HOST: 0.0.0.0
      MCP_PORT: 8000
      MCP_ALLOWED_HOSTS: 127.0.0.1:*,localhost:*,[::1]:*
      MCP_ALLOWED_ORIGINS: http://127.0.0.1:*,http://localhost:*,http://[::1]:*
    ports:
      - "127.0.0.1:${MCP_PUBLISH_PORT:-8000}:8000"
    restart: unless-stopped

ホストの8000番が使用中の場合は、MCP_PUBLISH_PORT=8766 docker compose up --buildのように公開ポートだけを変更し、クライアントのMCP_URLも同じポートへ合わせます。コンテナ内部の待受ポートは8000のままです。

起動はdocker compose up --buildです。コンテナ化によってPythonと依存パッケージ、起動方法は揃いますが、TLS、認証、秘密情報管理、監視、バックアップは追加されません。Docker公式も、コンテナをアプリケーションと依存関係をまとめる実行単位として説明しています。Docker Python guide

複数のクライアントから接続する

公式クライアントのstreamable_http_clientとClientSessionを使います。2つのセッションを同時に開き、Aがタスクを作成した後、Bが共有タスクを読み取ります。

import asyncio

from mcp import ClientSession
from mcp.client.streamable_http import streamable_http_client

URL = "http://127.0.0.1:8000/mcp"


async def client_a(created: asyncio.Event) -> None:
    async with streamable_http_client(URL) as (read, write, get_session_id):
        async with ClientSession(read, write) as session:
            await session.initialize()
            session_id = get_session_id()
            print("A session:", session_id[:8] if session_id else "stateless")
            result = await session.call_tool(
                "create_task",
                arguments={"title": "Review Streamable HTTP logs"},
            )
            print("A created:", result)
            created.set()
            await asyncio.sleep(0.5)


async def client_b(created: asyncio.Event) -> None:
    async with streamable_http_client(URL) as (read, write, get_session_id):
        async with ClientSession(read, write) as session:
            await session.initialize()
            session_id = get_session_id()
            print("B session:", session_id[:8] if session_id else "stateless")
            await created.wait()
            result = await session.call_tool("list_tasks", arguments={})
            print("B listed:", result)


async def main() -> None:
    created = asyncio.Event()
    await asyncio.gather(client_a(created), client_b(created))


if __name__ == "__main__":
    asyncio.run(main())

AとBに異なるセッションIDの先頭が表示され、Bの結果にAが作成したタスクが含まれれば、通信セッションが分離され、業務データが共有されていることを確認できます。ログへセッションIDを出す場合、本番ではID全体を無条件に記録せず、短縮値やハッシュを使います。

MCP InspectorでもtransportにStreamable HTTPを選び、http://127.0.0.1:8000/mcpへ接続できます。InspectorはTool一覧、入力スキーマ、結果、ログ、通知を対話的に確認するための公式ツールです。MCP Inspector

CORS、Origin、TLS、認証を分ける

これらは似た場所に登場しますが、役割は異なります。

CORSが必要なのは、主にブラウザ上のJavaScriptが別OriginのMCPサーバーへ直接接続する場合です。許可するOriginを限定し、GET、POST、DELETEおよび必要なリクエストヘッダーを許可します。Stateful構成では、ブラウザが初期化レスポンスのMcp-Session-Idを読めるよう、expose_headersへ追加する必要があります。公式SDKにはStarletteのCORSMiddlewareを使う例があります。Python SDK CORS guidance

ただし、CORSはブラウザがレスポンスをJavaScriptへ渡すかを決める仕組みです。curl、Python、別のバックエンドはCORSを強制しません。「CORSを設定したから認証は不要」は誤りです。

Origin検証はサーバー自身が接続を受け入れるか判断します。MCP仕様は、DNS rebinding対策として、存在するOriginヘッダーを検証し、不正なら403を返すことを求めています。本実装のTransportSecuritySettingsはHostも合わせて検証します。ブラウザ以外の正当なクライアントではOriginが付かない場合があるため、Originなしを一律拒否するのではなく、認証・ネットワーク境界と組み合わせます。

TLSはHTTPSによってTool入力、結果、セッションID、Authorizationヘッダーを通信経路上で保護します。localhost外へ出す本番構成では、リバースプロキシやロードバランサーでTLSを終端し、バックエンドまでの経路も信頼境界に応じて保護します。

認証なしでインターネットへ公開しない

MCP Toolはファイル更新、データベース操作、メール送信、チケット作成、クラウド操作、削除などの副作用を持てます。認証なしで公開すると、第三者はAIクライアントを経由せず、MCPプロトコルを実装した任意のプログラムからToolを呼び出せます。

次の仕組みは認証の代わりになりません。

• 推測しにくいURL
• MCPセッションID
• CORSまたはOrigin許可リスト
• HTTPS
• Docker
• 非公開のTool名
• Tool descriptionの注意書き
• AIモデルが表示する確認メッセージ

この記事で許可するのは、127.0.0.1限定のポート公開、閉じたDockerネットワーク、一時的な検証環境です。0.0.0.0でホスト外部へ公開すること、パブリックIPへ直接公開すること、認証なしでクラウドへデプロイすることは禁止です。

ユーザー固有データ、管理操作、監査、利用者別レート制限を扱うリモートMCPサーバーでは、認証・認可を設計します。MCPの公式ガイダンスは、HTTPベースのリモートサーバーについてOAuth 2.1の規約に沿った認可フローを説明しています。Understanding Authorization in MCP

まとめ

stdioからStreamable HTTPへの移行は、起動オプションの変更だけではありません。MCPサーバーを、クライアント管理下のローカルプロセスから、複数クライアントが利用する独立したネットワークサービスへ変える作業です。

Stateful構成では通信セッションを管理しますが、セッションは認証でも業務データでもありません。Streamable HTTPは必要に応じてSSEを使えますが、旧HTTP+SSE transportとは別の方式です。Docker内部では0.0.0.0へバインドしつつ、ホスト側は127.0.0.1だけに公開できます。

そして、CORS、Origin検証、TLS、認証、認可にはそれぞれ別の責任があります。Docker化しても本番対応が完了するわけではありません。認証を追加するまでは、MCPエンドポイントをlocalhostまたは閉じた検証環境に限定してください。