Client
Client 是 Python 程序与 MCP 服务器对话的方式。
它是一个对象，只有一套生命周期：构造它，进入 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)
顶部的服务器只是为了让你有东西可连。客户端就是高亮的那五行。
Client(mcp)接收的是服务器对象本身。这是内存传输：没有子进程，没有端口，没有 HTTP。本页的每个示例，以及你写的每个测试，都是这样连接的。async with就是生命周期。进入时连接并协商；离开时断开。没有connect()/close()这样的配对方法，而且Client在代码块结束后不能复用。- 在代码块内部，连接相关的信息已经作为普通属性摆在那里了。
可以传给 Client 什么
Client 接收一个位置参数，并根据它的类型确定传输方式：
MCPServer（或低层Server）实例：进程内连接。- URL 字符串（
Client("http://localhost:8000/mcp")）：Streamable HTTP，生产环境的路径。 - 传输：任何可以
async with ... as (read, write)的对象，比如包装子进程的stdio_client(...)。
本页其余内容在这三种方式下完全相同。请求头、子进程、超时以及 Transport 协议另有专页：客户端传输。
已连接的客户端上有什么
四个只读属性，进入代码块的那一刻就已填好：
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 会探测服务器，遇到较老的服务器就回退到经典握手，所以一个客户端能对接任何时代的服务器。需要控制这一点时，详见 协议版本。
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
一个返回值，三样东西要读。各自有不同的使用者。
content：模型读的内容
content 是一个内容块的 list，而内容块是一个联合类型：TextContent、ImageContent、AudioContent、ResourceLink 或 EmbeddedResource。一个工具可以返回多个不同种类的块。
这就是为什么 main 在碰 block.text 之前先用 isinstance(block, TextContent) 收窄类型。注意 isinstance 之外没有出现 .text：类型检查器不允许，因为 ImageContent 有的是 .data，不是 .text。这个联合类型如实表达了工具可以发给你什么；你的代码也应该如此。
structured_content：应用程序读的内容
structured_content 是工具返回值的 JSON 形式，符合工具声明的 output_schema。不用解析字符串，不用猜。
两者同时存在时，是有意把同一件事说两遍：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。只有当服务器回复的是 JSON-RPC 错误而不是结果时，Client 方法才会抛出 MCPError；服务器在什么情况下产生哪一种，见 处理错误。
资源
资源动词成对出现：两种列出方式，一种读取方式。
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}']。它们是两个不同的列表，因为模板在填好之前是不可读的。read_resource(uri)接收一个普通的strURI，对两者都适用：传入"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) 渲染它。参数字典是 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 一页返回全部内容，所以 next_cursor 是 None，循环只跑一次，这也是为什么大多数代码从来不写它。真正分页的服务器，以及游标遵守的规则，见 分页。
在测试中
没有进程、没有端口的 Client(mcp)，本身就是服务器的测试工具。
有一个构造参数专为此而设：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为止。
服务器可以向客户端请求什么，以及你如何回应，见 客户端回调。