Client
Client は、Python プログラムが MCP サーバーと対話するための手段です。
1 つのオブジェクトに 1 つのライフサイクルがあります。組み立てて、async with に入り、メソッドを呼び出します。プロトコルの動詞（ツールの一覧取得、ツールの呼び出し、リソースの読み取り、プロンプトのレンダリング）はどれも、このオブジェクトの async メソッドで、型付きの結果を返します。
最初のクライアント
from mcp import Client
from mcp.server import MCPServer
mcp = MCPServer("Bookshop", instructions="Search the catalog before recommending a book.")
@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:
print(client.server_info)
print(client.server_capabilities)
print(client.protocol_version)
print(client.instructions)
冒頭のサーバーは、接続先を用意するためだけにあります。クライアントはハイライトされた 5 行です。
Client(mcp)にはサーバーオブジェクトそのものを渡しています。これがインメモリのトランスポートです。サブプロセスもポートも HTTP もありません。このページのすべての例、そして作成するすべてのテストが、この方法で接続します。async withがライフサイクルです。入ると接続してネゴシエーションを行い、出ると切断します。connect()/close()のペアはなく、ブロックが終わった後のClientは再利用できません。- ブロックの中では、接続に関する情報がすでに通常のプロパティとして揃っています。
Client に渡せるもの
Client は位置引数を 1 つ取り、その型からトランスポートを決定します。
MCPServer（または低レベルのServer）のインスタンス：プロセス内で接続します。- URL 文字列（
Client("http://localhost:8000/mcp")）：Streamable HTTP。本番向けの経路です。 - トランスポート：
async with ... as (read, write)できるものなら何でも。たとえばサブプロセスをラップするstdio_client(...)です。
このページの残りの内容は、3 つのどれでも同じです。ヘッダー、サブプロセス、タイムアウト、そして Transport プロトコルについては、専用のページ クライアントのトランスポート があります。
接続済みクライアントが持つもの
読み取り専用のプロパティが 4 つあり、ブロックに入った瞬間に値が入ります。
client.server_info：サーバーの識別情報。報告しない 2026 年世代のサーバーではNoneです（python-sdk のサーバーはデフォルトで報告します）。ここではserver_info.nameが"Bookshop"で、server_info.versionはサーバーが報告する値です。client.server_capabilities：サーバーができること（tools、resources、prompts、completions、...）。サーバーが持たないケイパビリティはNoneです。client.protocol_version：両者が合意したプロトコルバージョン。ここでは"2026-07-28"です。client.instructions：サーバーのinstructions=文字列。設定されていなければNoneです。
プロトコルバージョンを選んだ覚えはないはずです。デフォルトでは Client がサーバーを調べ、古いサーバーに対しては従来のハンドシェイクにフォールバックします。そのため、1 つのクライアントがどの世代のサーバーに対しても動作します。これを制御する必要がある場合、詳しくは プロトコルバージョン を参照してください。
Tip
client.session は下層の ClientSession で、低レベルへの抜け道です。このページの内容では必要ありません。
ツールの一覧取得
from mcp import Client
from mcp.server import MCPServer
mcp = MCPServer("Bookshop")
@mcp.tool(title="Search the catalog")
def search_books(query: str, limit: int = 10) -> str:
"""Search the catalog by title or author."""
return f"Found 3 books matching {query!r} (showing up to {limit})."
async def main() -> None:
async with Client(mcp) as client:
result = await client.list_tools()
for tool in result.tools:
print(tool.name)
print(tool.title)
print(tool.description)
print(tool.input_schema)
list_tools() は ListToolsResult を返し、ツールは .tools に入っています。それぞれが、ホストがモデルに渡す完全な定義です。
tool.name # 'search_books'
tool.title # 'Search the catalog'
tool.description # 'Search the catalog by title or author.'
そして tool.input_schema は、サーバーが関数の型ヒントから導き出した JSON Schema です。
{
"type": "object",
"properties": {
"query": {"title": "Query", "type": "string"},
"limit": {"default": 10, "title": "Limit", "type": "integer"}
},
"required": ["query"],
"title": "search_booksArguments"
}
このスキーマには、UI が引数フォームを描画するのに必要なものも、モデルが有効な引数を生成するのに必要なものも、すべて含まれています。
Tip
title は省略可能なので、人間にツールを見せる UI はどちらかを選ぶ必要があります。title があればそれを、なければ name を使います。from mcp.shared.metadata_utils import get_display_name がまさにそれを行い、ツール、リソース、リソーステンプレート、プロンプトに対応しています。
ツールの呼び出し
call_tool(name, arguments) はツールを実行し、CallToolResult を返します。
from pydantic import BaseModel
from mcp import Client
from mcp.server import MCPServer
from mcp.types import TextContent
mcp = MCPServer("Bookshop")
class Book(BaseModel):
title: str
author: str
year: int
@mcp.tool()
def lookup_book(title: str) -> Book:
"""Look up a book by its exact title."""
if title != "Dune":
raise ValueError(f"No book titled {title!r} in the catalog.")
return Book(title="Dune", author="Frank Herbert", year=1965)
async def main() -> None:
async with Client(mcp) as client:
result = await client.call_tool("lookup_book", {"title": "Dune"})
for block in result.content:
if isinstance(block, TextContent):
print(block.text)
print(result.structured_content)
print(result.is_error)
サーバーの lookup_book は Pydantic の Book を返します。クライアントから見えるのは次のとおりです。
result.content # [TextContent(type='text', text='{\n "title": "Dune",\n "author": "Frank Herbert",\n "year": 1965\n}')]
result.structured_content # {'title': 'Dune', 'author': 'Frank Herbert', 'year': 1965}
result.is_error # False
戻り値は 1 つ、読むべきものは 3 つです。それぞれ読み手が異なります。
content：モデルが読むもの
content はコンテンツブロックの list で、コンテンツブロックはユニオン型です。TextContent、ImageContent、AudioContent、ResourceLink、EmbeddedResource のいずれかです。ツールは種類の異なるブロックを複数返せます。
main が block.text に触れる前に isinstance(block, TextContent) で絞り込んでいるのはそのためです。isinstance の外に .text がないことに注目してください。ImageContent が持つのは .text ではなく .data なので、型チェッカーが許しません。このユニオンは、ツールが送ってよいものを正直に表しています。コードもそうあるべきです。
structured_content：アプリケーションが読むもの
structured_content はツールの戻り値を JSON にしたもので、ツールが宣言した output_schema に一致します。文字列の解析も推測も不要です。
両方があるときは、意図的に同じことを 2 回言っています。content はモデル向け、structured_content はコード向けです。構造化されたほうがどこから来るのか、どう制御するのかは、構造化出力 のページで説明しています。
is_error：ツールが失敗したかどうか
例外を送出するツールが、クライアント側で例外を送出することはありません。is_error=True の付いた通常の結果として返ってきます。
Check
lookup_book に "Solaris"（カタログにない書名）を問い合わせると、関数は ValueError を送出します。それでも呼び出しは正常に返ります。
result.is_error # True
result.content # [TextContent(type='text', text="Error executing tool lookup_book: No book titled 'Solaris' in the catalog.")]
result.structured_content # None
例外のメッセージは content に入りました。そこならモデルが読んで、やり直せます。これは意図的なものです。ツールのエラーはクラッシュではなく、会話の一部です。structured_content を信用する前に、必ず is_error を確認してください。
Warning
is_error=True がカバーするのは、自分で書いた raise だけではありません。サーバーに存在すらしないツールを要求しても（call_tool("does_not_exist", {})）、何も送出されません。同じ形の結果が返り、is_error=True で content には Unknown tool: does_not_exist が入ります。Client のメソッドが MCPError を送出するのは、サーバーが結果ではなく JSON-RPC のエラーで応答したときだけです。サーバーがどんなときにどちらを返すかは エラーの処理 で扱っています。
リソース
リソースの動詞は組になっています。一覧取得が 2 通り、読み取りが 1 通りです。
from mcp import Client
from mcp.server import MCPServer
from mcp.types import TextResourceContents
mcp = MCPServer("Bookshop")
@mcp.resource("catalog://genres")
def genres() -> list[str]:
"""The genres the catalog is organised by."""
return ["fiction", "non-fiction", "poetry"]
@mcp.resource("catalog://genres/{genre}")
def books_in_genre(genre: str) -> str:
"""Every title we stock in one genre."""
return f"3 books filed under {genre}."
async def main() -> None:
async with Client(mcp) as client:
listed = await client.list_resources()
print([resource.uri for resource in listed.resources])
templates = await client.list_resource_templates()
print([template.uri_template for template in templates.resource_templates])
result = await client.read_resource("catalog://genres/poetry")
for contents in result.contents:
if isinstance(contents, TextResourceContents):
print(contents.text)
list_resources()は具体的なリソース、つまり URI が固定のものを返します。ここでは['catalog://genres']です。list_resource_templates()はパラメーター化されたものを返します。ここでは['catalog://genres/{genre}']です。テンプレートは値を埋めるまで読み取れないため、2 つは別々のリストになっています。read_resource(uri)は通常のstrの URI を受け取り、両方に対して動作します。"catalog://genres/poetry"を渡せば、サーバーがテンプレートに照合します。
read_resource は contents を返します。これは TextResourceContents または BlobResourceContents のリストです。考え方はツールのコンテンツと同じで、isinstance で絞り込んでから .text（または .blob）を読みます。
クライアントは、リソースが変更されたときに通知を受けることもできます。2025 年世代の接続では subscribe_resource(uri) / unsubscribe_resource(uri) がそれにあたります。ただしこのメソッドのペアは MCPServer が実装していないため、2026-07-28 の通信上（これらの動詞はもう存在しません）ではリクエストに -32601、Method not found が返ります。2026 年の代替は subscriptions/listen ストリームで、こちらは MCPServer が実際に提供しています（そこでは server_capabilities.resources.subscribe が True です）。これを client.listen(...) で消費する方法は、このセクションの サブスクリプション のページで説明しています。
プロンプト
from mcp import Client
from mcp.server import MCPServer
mcp = MCPServer("Bookshop")
@mcp.prompt(title="Recommend a book")
def recommend(genre: str) -> str:
"""Ask for a recommendation in a genre."""
return f"Recommend one {genre} book from the catalog and say why."
async def main() -> None:
async with Client(mcp) as client:
listed = await client.list_prompts()
print(listed.prompts)
result = await client.get_prompt("recommend", {"genre": "poetry"})
for message in result.messages:
print(message.role, message.content)
list_prompts() は、サーバーが何を提供していて、各プロンプトが何を必要とするかを教えてくれます。
prompt.name # 'recommend'
prompt.title # 'Recommend a book'
prompt.arguments # [PromptArgument(name='genre', required=True)]
get_prompt(name, arguments) でレンダリングします。引数の dict は str -> str で、プロンプトの引数は常に文字列です。結果は messages、つまり PromptMessage のリストで、それぞれが role と content ブロックを持ちます。
message.role # 'user'
message.content # TextContent(type='text', text='Recommend one poetry book from the catalog and say why.')
ホストはこれらのメッセージをそのままモデルに渡します。機能はこれだけです。
補完
補完ハンドラーを持つサーバーは、ユーザーの入力に合わせてプロンプトやリソーステンプレートの引数を自動補完できます。
from mcp import Client
from mcp.server import MCPServer
from mcp.types import Completion, CompletionArgument, CompletionContext, PromptReference, ResourceTemplateReference
mcp = MCPServer("Bookshop")
GENRES = ["fiction", "non-fiction", "poetry"]
@mcp.prompt()
def recommend(genre: str) -> str:
"""Ask for a recommendation in a genre."""
return f"Recommend one {genre} book from the catalog and say why."
@mcp.completion()
async def complete_genre(
ref: PromptReference | ResourceTemplateReference,
argument: CompletionArgument,
context: CompletionContext | None,
) -> Completion | None:
return Completion(values=[genre for genre in GENRES if genre.startswith(argument.value)])
async def main() -> None:
async with Client(mcp) as client:
result = await client.complete(
ref=PromptReference(type="ref/prompt", name="recommend"),
argument={"name": "genre", "value": "p"},
)
print(result.completion.values)
refは、「どの」プロンプトまたはテンプレートを埋めているかを示します。PromptReferenceまたはResourceTemplateReferenceです。argumentは{"name": ..., "value": ...}で、引数と、ユーザーがこれまでに入力した内容です。
答えは result.completion.values に入っています。"p" と入力すると、サーバーは ['poetry'] を返します。サーバー側の実装と、ハンドラーがすでに埋まっている「他の」引数を使って候補を絞り込む方法は、補完 のページで説明しています。
ページネーション
list_* メソッドはどれも cursor= キーワードを取り、結果はどれも next_cursor を持ちます。next_cursor が None なら、すべて取得済みです。
from mcp import Client
from mcp.server import MCPServer
from mcp.types import Tool
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}."
@mcp.tool()
def reserve_book(title: str) -> str:
"""Put a book on hold."""
return f"Reserved {title!r}."
async def main() -> None:
async with Client(mcp) as client:
tools: list[Tool] = []
cursor: str | None = None
while True:
page = await client.list_tools(cursor=cursor)
tools.extend(page.tools)
if page.next_cursor is None:
break
cursor = page.next_cursor
print([tool.name for tool in tools])
このループはどのサーバーに対しても正しく動きます。MCPServer はすべてを 1 ページで返すので、next_cursor は None になり、ループは 1 回だけ実行されます。ほとんどのコードがこのループを書かないのはそのためです。実際にページ分割するサーバーと、カーソルが従うルールについては ページネーション を参照してください。
テストでの利用
プロセスもポートも使わない Client(mcp) は、それだけでサーバーのテストハーネスになります。
そのために用意されたコンストラクターのフラグが 1 つあります。Client(mcp, raise_exceptions=True) です。効果があるのはインメモリ接続のときだけで、その説明と、それを中心にしたパターン全体の組み立ては テスト のページにあります。
まとめ
Client(x)は、サーバーオブジェクトにはインメモリで、URL 文字列には Streamable HTTP で、それ以外にはトランスポート経由で接続します。async withがライフサイクルのすべてです。その中ではserver_capabilitiesとprotocol_versionにすでに値が入っており、サーバーが提供していればserver_infoとinstructionsも同様です。list_tools()で各ツールのname、title、description、input_schemaが得られます。call_tool()はモデル向けのcontent、コード向けのstructured_content、そしてis_errorを返します。例外を送出するツールは、例外ではなく結果として返ってきます。contentはブロック型のユニオンです。読む前にisinstanceで絞り込みます。list_resources/list_resource_templates/read_resource、list_prompts/get_prompt、そしてcompleteで動詞は一通り揃います。list_*はどれもcursor=を取ります。next_cursorがNoneになるまでループします。
サーバーのほうからクライアントに要求できることと、それにどう応えるかは、クライアントのコールバック で扱います。