クライアントのトランスポート

どの Client も、トランスポートを介してサーバーと対話します。トランスポートとは、実際にメッセージを運ぶもののことです。

トランスポートを別途設定することはありません。Client は位置引数を 1 つだけ受け取り、その型からトランスポートを判断します。

それぞれの「サーバー」側（mcp.run() が何をするのか、何をデプロイするのか）については、サーバーの実行 を参照してください。

インメモリ

サーバーオブジェクトそのものを渡します。

from mcp import Client
from mcp.server import MCPServer

mcp = MCPServer("Bookshop")


@mcp.tool()
def search_books(query: str) -> str:
    """Search the catalog by title or author."""
    return f"Found 3 books matching {query!r}."


async def main() -> None:
    async with Client(mcp) as client:
        result = await client.call_tool("search_books", {"query": "dune"})
        print(result.structured_content)

サブプロセスも、ポートも、通信路を流れるバイト列もありません。クライアントとサーバーは同じプロセス内の 2 つのオブジェクトですが、呼び出しは本物のプロトコル層を通ります。search_books は、HTTP 越しの場合とまったく同じように一覧に載り、検証され、呼び出されます。

そのため、これは同時に 2 つの役割を果たします。

• テストハーネス。 このドキュメントの例はすべてこの方法で実行されており、テスト のページはこのパターンを中心に組み立てられています。
• 組み込み用の API。 サーバーを自分で構築するアプリケーションなら、そのツールを呼び出すのにネットワーク越しの経路は必要ありません。

Streamable HTTP

URL の文字列を渡すと Streamable HTTP になります。デプロイ時に使うトランスポートです。

from mcp import Client


async def main() -> None:
    async with Client("http://localhost:8000/mcp") as client:
        result = await client.list_tools()
        print([tool.name for tool in result.tools])

本番用のクライアントはこれですべてです。Client は URL を streamable_http_client(...) で包み、MCP に必要な設定を施した httpx2.AsyncClient の上に載せてくれます。具体的には follow_redirects=True、connect/write/pool のタイムアウトが 30 秒、そしてサーバーがレスポンスストリームを開いたままにすることがあるため read のタイムアウトが 300 秒です。

RuntimeError: Client must be used within an async context manager

自前の httpx2.AsyncClient を使う

Authorization ヘッダー、Cookie、プロキシ、mTLS、あるいは別のタイムアウトが必要になったら、httpx2.AsyncClient を自分で組み立てて streamable_http_client に渡します。

import httpx2

from mcp import Client
from mcp.client.streamable_http import streamable_http_client


async def main() -> None:
    async with httpx2.AsyncClient(
        headers={"Authorization": "Bearer ..."},
        timeout=httpx2.Timeout(30.0, read=300.0),
        follow_redirects=True,
    ) as http_client:
        transport = streamable_http_client("http://localhost:8000/mcp", http_client=http_client)
        async with Client(transport) as client:
            result = await client.list_tools()
            print([tool.name for tool in result.tools])

注目すべき点が 2 つあります。

• httpx2.AsyncClient の所有者は自分なので、入るのも出るのも自分で行います。SDK は自身が作成していないクライアントを決して閉じません。
• streamable_http_client(url, http_client=...) はトランスポートを返し、Client(transport) はそれを他のものと同じように受け取ります。

TLS について 1 点。httpx2 は、同梱の CA リストではなく、オペレーティングシステムのトラストストアに対して証明書を検証します（truststore を使用）。利用できるシステム CA ストアがない環境（一部の最小構成コンテナなど）では、標準の環境変数 SSL_CERT_FILE/SSL_CERT_DIR を設定するか、httpx2.AsyncClient に明示的に verify=ssl_context を渡してください（背景は httpx と httpx-sse の httpx2 への置き換えを参照）。

TypeError: streamable_http_client() got an unexpected keyword argument 'headers'

stdio

stdio サーバーはサブプロセスです。クライアントがそれを起動し、stdin に JSON-RPC を書き込み、stdout から JSON-RPC を読み取ります。デスクトップのホストが手元のマシンでサーバーを動かす方法がこれです。ホストとは、このコードに UI を加えたもの「そのもの」です。本物のホストに接続する は、同じ関係をホストの側から設定ファイルとして見たものです。

StdioServerParameters でプロセスを記述し、stdio_client でトランスポートに変換して、「それ」を Client に渡します。

from mcp import Client, StdioServerParameters
from mcp.client.stdio import stdio_client

server = StdioServerParameters(
    command="uv",
    args=["run", "server.py"],
    env={"BOOKSHOP_API_KEY": "secret"},
)


async def main() -> None:
    async with Client(stdio_client(server)) as client:
        result = await client.list_tools()
        print([tool.name for tool in result.tools])

Client はパラメーターオブジェクトをそのままでは受け取りません。StdioServerParameters は設定であり、stdio_client(server) はそこからプロセスを起動する方法を知っているトランスポートです。必ず包んでください。

async with ブロックを抜けると、サブプロセスも終了されます。stdin を閉じ、待機し、居残っていれば強制終了します。自分で後始末をすることはありません。

SSE

mcp.client.sse の sse_client(url) は、Streamable HTTP に取って代わられた HTTP トランスポートです。まだこれを話すサーバーと対話するには、同じように Client(sse_client("http://localhost:8000/sse")) と包みます。そして、新しいものをこの上に作らないでください。

Transport プロトコル

Client から見れば、上記はすべて同じものです。

トランスポートとは、(read, write) というメッセージストリームのペアを yield する非同期コンテキストマネージャーのことです。正式には mcp.client の Transport プロトコルです。Client は引数を型で解決します。サーバーオブジェクトならインプロセスで接続し、str なら streamable_http_client(url) になり、それ以外は直接トランスポートとして入ります。この最後の規則があるからこそ、stdio_client(...)、streamable_http_client(...)、sse_client(...) はすべて同じ場所に収まり、自分で独自のものを書くこともできます。

まとめ

• Client(mcp)（サーバーオブジェクト）はインメモリで接続します。テストと組み込みに使ってください。
• Client("http://.../mcp")（URL）は、本番用のトランスポートである Streamable HTTP で接続します。
• ヘッダー、認証、プロキシ、タイムアウトは、streamable_http_client(url, http_client=...) に渡す httpx2.AsyncClient に設定します。headers= キーワードはありません。
• stdio は Client(stdio_client(StdioServerParameters(...))) であり、パラメーターオブジェクト単体では決してありません。
• サブプロセスが受け取るのは自分の環境ではなく、許可リストに基づく環境です。env= でそこに追加します。
• トランスポートとは、async with x as (read, write) と書けるものすべてです。Client は、サーバーオブジェクトでも URL でもないものをそのままこのプロトコルに渡します。
• Client の構築でトランスポートが選ばれ、async with でそれが開かれます。

トランスポートが開いたら、両者はプロトコルバージョンについて合意する必要があります。普段は意識することはありません。意識することになったら、プロトコルバージョン のページを参照してください。