跳轉至

用戶端

機器翻譯

本頁是從英文說明文件自動翻譯而來,以英文頁面為準。如果哪裡讀起來不對勁,翻譯有說明如何回報。

Python 程式要和 MCP 伺服器對話,靠的就是 Client

它是一個物件,只有一套生命週期:建立它、進入 async with、呼叫方法。每個協定動作(列出工具、呼叫其中一個、讀取資源、算繪提示詞)都是它上面的一個 async 方法,回傳有型別的結果。

你的第一個用戶端

用戶端需要有伺服器可以對話。這一頁的每段程式碼連的都是這間 Bookshop。把它存成 server.py,讓它透過 HTTP 持續執行:

server.py
from pydantic import BaseModel

from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.types import Completion, CompletionArgument, CompletionContext, PromptReference, ResourceTemplateReference

mcp = MCPServer("Bookshop", instructions="Search the catalog before recommending a book.")

GENRES = ["fiction", "non-fiction", "poetry"]


class Book(BaseModel):
    title: str
    author: str
    year: int


@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})."


@mcp.tool()
def lookup_book(title: str) -> Book:
    """Look up a book by its exact title."""
    if title != "Dune":
        raise ToolError(f"No book titled {title!r} in the catalog.")
    return Book(title="Dune", author="Frank Herbert", year=1965)


@mcp.resource("catalog://genres")
def genres() -> list[str]:
    """The genres the catalog is organised by."""
    return GENRES


@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}."


@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."


@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)])
uv run mcp run server.py --transport streamable-http

這樣伺服器就在 http://localhost:8000/mcp 提供服務。用戶端是另一個獨立的程式。把它存成 client.py,在第二個終端機執行 python client.py

client.py
import anyio

from mcp import Client


async def main() -> None:
    async with Client("http://localhost:8000/mcp") as client:
        print(client.server_info)
        print(client.server_capabilities)
        print(client.protocol_version)
        print(client.instructions)


if __name__ == "__main__":
    anyio.run(main)
  • Client("http://localhost:8000/mcp") 拿到的是一個 URL,所以它透過 Streamable HTTP 連到你剛啟動的伺服器。
  • async with 就是生命週期。進入時連線並協商;離開時斷線。沒有 connect() / close() 這種成對的方法,而且區塊結束後 Client 不能再重複使用。
  • 在區塊內,連線的各項資訊已經以普通屬性的形式準備好了。

可以傳什麼給 Client

Client 接受一個位置引數,並依它的型別決定傳輸方式:

  • URL 字串(Client("http://localhost:8000/mcp")):Streamable HTTP,也就是部署時使用的傳輸方式。
  • StdioServerParameters:要當作本機子處理程序啟動的命令,透過它的 stdin 和 stdout 溝通。
  • 一個傳輸:任何可以 async with ... as (read, write) 的東西,例如用 streamable_http_client(url, http_client=...) 包住你自己的 HTTP 用戶端。
  • MCPServer(或低階的 Server)實例:在同一個處理程序內連線,沒有子處理程序,也沒有連接埠。這一種是給測試用的,測試 就是以它為基礎。

這一頁其餘的內容在這四種情況下完全相同。標頭、子處理程序、逾時,以及 Transport 協定另外有專屬的頁面:用戶端傳輸方式

連線後的用戶端上有什麼

四個唯讀屬性,一進入區塊就填好了:

  • client.server_info:伺服器的身分;如果是不回報身分的 2026 世代伺服器,則為 None(python-sdk 伺服器預設會回報)。這裡的 server_info.name"Bookshop"server_info.version 則是伺服器回報的值。
  • client.server_capabilities:伺服器能做什麼(toolsresourcespromptscompletions……)。伺服器沒有的能力會是 None
  • client.protocol_version:雙方談妥的協定版本。這裡是 "2026-07-28"
  • client.instructions:伺服器的 instructions= 字串;如果沒有設定則為 None

你從頭到尾都沒有挑過協定版本。預設情況下 Client 會先探測伺服器,遇到較舊的伺服器就退回傳統的交握,所以同一個用戶端可以對應任何世代的伺服器。需要自己掌控時,完整說明請見 協定版本

Tip

client.session 是底層的 ClientSession,也就是低階的逃生出口。這一頁的任何內容都用不到它。

列出工具

client.py
import anyio

from mcp import Client


async def main() -> None:
    async with Client("http://localhost:8000/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)


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

list_tools() 回傳一個 ListToolsResult;工具在 .tools 裡。每一個都是 MCP 主機(host)會交給模型的完整定義。第一個是這樣:

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"
}

這份 schema 就是 UI 算繪引數表單所需的一切,也是模型產生合法引數所需的一切。

第二個工具 lookup_book 註冊時沒有給 title=,所以它的 tool.titleNone

Tip

title 是選填的,所以把工具顯示給人看的 UI 得自己挑:有的話就用 title,沒有就用 namefrom mcp.shared.metadata_utils import get_display_name 做的正是這件事,適用於工具、資源、資源範本和提示詞。

呼叫工具

call_tool(name, arguments) 會執行工具,並回傳一個 CallToolResult

client.py
import anyio

from mcp import Client
from mcp.types import TextContent


async def main() -> None:
    async with Client("http://localhost:8000/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)


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

伺服器的 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,而內容區塊是一個聯集:TextContentImageContentAudioContentResourceLinkEmbeddedResource。一個工具可以回傳好幾個,而且種類各異。

這就是為什麼 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"(目錄裡沒有的書名),函式會引發 ToolError。呼叫仍然正常回傳:

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

ToolError 的訊息落在 content 裡,模型可以讀到它並再試一次。這是刻意的設計:工具錯誤是對話的一部分,不是當機。(假如工具是因為其他例外而當掉,content 就只會寫 Error executing tool lookup_book。)在信任 structured_content 之前,一定要先看 is_error

Warning

is_error=True 涵蓋的不只是你自己的 raise。要一個伺服器根本沒有的工具(call_tool("does_not_exist", {})),也不會引發任何例外。你會拿回同樣的形狀:is_error=Truecontent 裡是 Unknown tool: does_not_exist。只有在伺服器回的是 JSON-RPC 錯誤而不是結果時,Client 的方法才會引發 MCPError;伺服器什麼時候產生哪一種,請見 處理錯誤

資源

資源的動作是成組的:兩種列出的方式,一種讀取的方式。

client.py
import anyio

from mcp import Client
from mcp.types import TextResourceContents


async def main() -> None:
    async with Client("http://localhost:8000/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)


if __name__ == "__main__":
    anyio.run(main)
  • list_resources() 回傳具體的資源,也就是 URI 固定的那些。這裡是:['catalog://genres']
  • list_resource_templates() 回傳參數化的那些。這裡是:['catalog://genres/{genre}']。它們分成兩個清單,因為範本在填好之前是不能讀的。
  • read_resource(uri) 接受一個普通的 str URI,兩種都適用:傳入 "catalog://genres/poetry",伺服器會把它比對到範本。

read_resource 回傳 contents,一個由 TextResourceContentsBlobResourceContents 組成的清單。跟工具內容是同樣的概念:用 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.subscribeTrue——而用 client.listen(...) 來消費它,就是本節的 訂閱 頁面。

提示詞

client.py
import anyio

from mcp import Client


async def main() -> None:
    async with Client("http://localhost:8000/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)


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

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.')

主機會把這些訊息直接交給模型。整個功能就這樣。

自動完成

有自動完成處理函式的伺服器,可以在使用者輸入時自動完成提示詞和資源範本的引數。

client.py
import anyio

from mcp import Client
from mcp.types import PromptReference


async def main() -> None:
    async with Client("http://localhost:8000/mcp") as client:
        result = await client.complete(
            ref=PromptReference(type="ref/prompt", name="recommend"),
            argument={"name": "genre", "value": "p"},
        )
        print(result.completion.values)


if __name__ == "__main__":
    anyio.run(main)
  • ref 指出你正在填的是哪一個提示詞或範本:PromptReferenceResourceTemplateReference
  • argument{"name": ..., "value": ...}:引數本身,以及使用者目前為止輸入的內容。

答案在 result.completion.values 裡。輸入 "p",伺服器回的是 ['poetry']。伺服器端的部分,以及處理函式如何利用其他已填好的引數來縮小建議範圍,請見 自動完成 頁面。

分頁

每個 list_* 方法都接受 cursor= 關鍵字引數,每個結果都帶有 next_cursor。當 next_cursorNone,表示全部拿到了。

client.py
import anyio

from mcp import Client
from mcp.types import Tool


async def list_all_tools(client: Client) -> list[Tool]:
    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:
            return tools
        cursor = page.next_cursor


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


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

list_all_tools 對任何伺服器都正確。MCPServer 會一頁回傳全部,所以 next_cursorNone,迴圈只跑一次,這也是為什麼大部分程式碼從來不寫它。真正會分頁的伺服器,以及游標遵守的規則,請見 分頁

在測試中

這一頁的每個 client.py 都是透過 HTTP 連到 server.py。在測試裡可以跳過網路,直接把伺服器物件本身交給 Clientfrom server import mcp,然後 Client(mcp)。沒有處理程序、沒有連接埠,而上面的每個方法用起來都一樣。

有一個建構子旗標是專為此設計的:Client(mcp, raise_exceptions=True)。它只對同一處理程序內的連線有作用,而 測試 頁面會解釋它,並圍繞它建立整套模式。

重點回顧

  • Client(x) 對 URL 字串透過 Streamable HTTP 連線,對 StdioServerParameters 啟動子處理程序,對傳輸則直接進入,在測試中則接受伺服器物件本身。
  • async with 就是整個生命週期。在裡面,server_capabilitiesprotocol_version 已經填好;伺服器有提供時,server_infoinstructions 也是。
  • list_tools() 給你每個工具的 nametitledescriptioninput_schema
  • call_tool() 回傳給模型的 content、給程式碼的 structured_content,以及 is_error。會引發例外的工具是一個結果,不是例外。
  • content 是區塊型別的聯集;讀取前先用 isinstance 縮窄。
  • list_resources / list_resource_templates / read_resourcelist_prompts / get_prompt,以及 complete 補齊了其餘的動作。
  • 每個 list_* 都接受 cursor=;一直迴圈到 next_cursorNone 為止。

伺服器可以向用戶端要求的東西,以及你如何回應,請見 用戶端回呼