低階 Server
@mcp.tool() 是一層包裝。底下還有第二個伺服器類別 Server，講的是原始的 MCP：把協定物件交給它，它就原封不動地放上線路。
MCPServer 就是建構在它之上。當便利層礙事時，才往下走：
- 需要送出精確的 schema（從檔案載入、從資料庫產生），而不是從 Python 簽章推導出來的。
- 需要完全掌控結果：
_meta、is_error、structured_content的每一個鍵。 - 需要處理 MCP 沒有定義的方法。
其他情況，就留在 MCPServer。
同一個工具，手工打造
這是 工具 用九行 @mcp.tool() 寫出的 search_books 工具，拿掉語法糖之後的樣子：
server.py
from mcp.server import Server, ServerRequestContext
from mcp.types import (
CallToolRequestParams,
CallToolResult,
ListToolsResult,
PaginatedRequestParams,
TextContent,
Tool,
)
SEARCH_BOOKS = Tool(
name="search_books",
description="Search the catalog by title or author.",
input_schema={
"type": "object",
"properties": {"query": {"type": "string"}, "limit": {"type": "integer"}},
"required": ["query", "limit"],
},
)
async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
return ListToolsResult(tools=[SEARCH_BOOKS])
async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
args = params.arguments or {}
text = f"Found 3 books matching {args['query']!r} (showing up to {args['limit']})."
return CallToolResult(content=[TextContent(type="text", text=text)])
server = Server("Bookshop", on_list_tools=list_tools, on_call_tool=call_tool)
改了三件事，而這三件事就是整個低階 API：
- 處理函式是建構子參數。
on_list_tools=和on_call_tool=傳進Server(...)。這一層沒有裝飾器，而且每個處理函式的形狀都一樣：async (ctx, params) -> result。 - 輸入 schema 自己寫。
Tool.input_schema是普通的 JSON Schemadict。沒有人會從型別提示推導它，因為根本沒有型別提示可以推導。 - 結果自己組。
CallToolResult(content=[TextContent(...)])，手動建立。沒有任何東西會被包裝、轉換，或從回傳註記推斷出來。
params 是解析後的請求：CallToolRequestParams 提供 .name 和 .arguments。ctx 是 ServerRequestContext：ctx.session 用來回頭和用戶端溝通，還有 ctx.lifespan_context、ctx.request_id，以及 ctx.meta，也就是請求傳入的 _meta。
Info
如果用過 FastAPI，這個關係你早就認識了。MCPServer 是裝飾器加型別提示的那一層；Server 是底下的 Starlette。兩者不是競爭對手：MCPServer 會建立一個 Server，並在上面註冊和這些一模一樣的處理函式。
試試看
這個沒有 Inspector 可用：mcp dev 和 mcp run 只接受 MCPServer。記憶體內的 Client 則不在乎；它接收低階 Server 的方式和接收 MCPServer 完全一樣：
main.py
import asyncio
from mcp import Client
from server import server
async def main() -> None:
async with Client(server) as client:
result = await client.call_tool("search_books", {"query": "dune", "limit": 5})
print(result.content)
asyncio.run(main())
[TextContent(type='text', text="Found 3 books matching 'dune' (showing up to 5).", annotations=None, meta=None)]
和 @mcp.tool() 版本產生的文字一模一樣。坦白說有兩個差異：
result.structured_content是None。高階伺服器會幫你把-> str包成{"result": ...}；在這裡，你沒建的東西，沒有人會替你建。list_tools回傳的是你打出來的 schema，一字不差。高階版本每個屬性上都有"title": "Query"，根部還有一個"title": "search_booksArguments"：那是 Pydantic 的產物。在這一層，線路上有的東西，都是你放上去的。
沒有人替你檢查
MCPServer 會在函式執行之前就拒絕錯誤的引數，依照它產生的 schema 驗證這次呼叫（工具）。
Server 不做這件事。你的 input_schema 是公告給用戶端看的；從來不會套用到 params.arguments 上。
Check
呼叫 search_books 時不帶 limit，args["limit"] 就會引發 KeyError。用戶端看到的是：
MCPError: Internal server error
一個 JSON-RPC 錯誤，錯誤碼 -32603，訊息刻意寫得很籠統：SDK 不會把你的 traceback 洩漏給遠端呼叫端。模型永遠不知道自己哪裡做錯，所以無法重試。（在測試中，raise_exceptions=True 會改為浮現真正的例外；請見 測試。）
這可以推而廣之。從低階處理函式引發的例外永遠是協定錯誤，絕不會是 is_error=True 的工具結果。如果希望模型讀到失敗並恢復，就自己驗證 params.arguments，然後回傳 CallToolResult(content=[TextContent(...)], is_error=True)。這兩種失敗正是 處理錯誤 的主題。
兩個工具，一個處理函式
on_call_tool 是伺服器上所有工具唯一的進入點。依 params.name 分派：
server.py
from mcp.server import Server, ServerRequestContext
from mcp.types import (
CallToolRequestParams,
CallToolResult,
ListToolsResult,
PaginatedRequestParams,
TextContent,
Tool,
)
SEARCH_BOOKS = Tool(
name="search_books",
description="Search the catalog by title or author.",
input_schema={
"type": "object",
"properties": {"query": {"type": "string"}, "limit": {"type": "integer"}},
"required": ["query", "limit"],
},
)
ADD_BOOK = Tool(
name="add_book",
description="Add a book to the catalog.",
input_schema={
"type": "object",
"properties": {"title": {"type": "string"}, "author": {"type": "string"}, "year": {"type": "integer"}},
"required": ["title", "author", "year"],
},
)
async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
return ListToolsResult(tools=[SEARCH_BOOKS, ADD_BOOK])
async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
args = params.arguments or {}
if params.name == "search_books":
text = f"Found 3 books matching {args['query']!r} (showing up to {args['limit']})."
elif params.name == "add_book":
text = f"Added {args['title']!r} by {args['author']} ({args['year']})."
else:
raise ValueError(f"Unknown tool: {params.name}")
return CallToolResult(content=[TextContent(type="text", text=text)])
server = Server("Bookshop", on_list_tools=list_tools, on_call_tool=call_tool)
list_tools公告兩者。call_tool依名稱分派。else分支很重要：就算是你從沒列出過的名稱，Server也會照樣把它的tools/call直接轉進你的處理函式。在那裡引發例外，這次呼叫就會變成和上面一樣的-32603。
結構化輸出，手工打造
在 Tool 上宣告 output_schema，並在結果上放 structured_content。兩者都由你負責：
server.py
from mcp.server import Server, ServerRequestContext
from mcp.types import (
CallToolRequestParams,
CallToolResult,
ListToolsResult,
PaginatedRequestParams,
TextContent,
Tool,
)
SEARCH_BOOKS = Tool(
name="search_books",
description="Search the catalog by title or author.",
input_schema={
"type": "object",
"properties": {"query": {"type": "string"}, "limit": {"type": "integer"}},
"required": ["query", "limit"],
},
output_schema={
"type": "object",
"properties": {"matches": {"type": "integer"}, "query": {"type": "string"}},
"required": ["matches", "query"],
},
)
async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
return ListToolsResult(tools=[SEARCH_BOOKS])
async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
args = params.arguments or {}
data = {"matches": 3, "query": args["query"]}
return CallToolResult(
content=[TextContent(type="text", text=f"Found 3 books matching {args['query']!r}.")],
structured_content=data,
)
server = Server("Bookshop", version="2.0.0", on_list_tools=list_tools, on_call_tool=call_tool)
呼叫它，結果會同時帶著兩種表示法：
{
"content": [{"type": "text", "text": "Found 3 books matching 'dune'."}],
"structuredContent": {"matches": 3, "query": "dune"},
"isError": false,
"resultType": "complete",
"_meta": {"io.modelcontextprotocol/serverInfo": {"name": "Bookshop", "version": "2.0.0"}}
}
_meta 區塊是伺服器的身分戳記：SDK 會把它加到每個 2026 世代的結果上，version 取自建構子（沒設定的伺服器會回報空字串）。不能表明身分的伺服器可以用中介軟體把這個鍵拿掉，中介軟體擁有它回傳的結果。
伺服器從不比對這兩個欄位。這個 SDK 的 Client 會：回傳的 structured_content 如果不符合你宣告的 output_schema，call_tool 就會引發 RuntimeError，訊息以 Invalid structured content returned by tool search_books 開頭，接著引用 jsonschema 的失敗內容。承諾一個 schema 很便宜；守住承諾是你的事。回傳型別與 schema 的完整階梯請見 結構化輸出。
_meta：給應用程式，不是給模型
content 是答案中模型會讀的部分。structured_content 是同一個答案的型別化資料。_meta 是第三個管道：跟著結果一起送給用戶端應用程式的資料，完全不屬於答案的一部分。
用它放紀錄 ID、追蹤 ID，任何 UI 需要而提示詞不需要的東西：
server.py
from mcp.server import Server, ServerRequestContext
from mcp.types import (
CallToolRequestParams,
CallToolResult,
ListToolsResult,
PaginatedRequestParams,
TextContent,
Tool,
)
SEARCH_BOOKS = Tool(
name="search_books",
description="Search the catalog by title or author.",
input_schema={
"type": "object",
"properties": {"query": {"type": "string"}, "limit": {"type": "integer"}},
"required": ["query", "limit"],
},
output_schema={
"type": "object",
"properties": {"matches": {"type": "integer"}, "query": {"type": "string"}},
"required": ["matches", "query"],
},
)
async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
return ListToolsResult(tools=[SEARCH_BOOKS])
async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
args = params.arguments or {}
data = {"matches": 3, "query": args["query"]}
return CallToolResult(
content=[TextContent(type="text", text=f"Found 3 books matching {args['query']!r}.")],
structured_content=data,
_meta={"bookshop/record_ids": ["bk_17", "bk_42", "bk_99"]},
)
server = Server("Bookshop", on_list_tools=list_tools, on_call_tool=call_tool)
- 建構時寫成
_meta=，也就是線路上的名稱。用戶端讀回來時是result.meta。 - 替鍵加上命名空間（
bookshop/record_ids）。io.modelcontextprotocol/*這些鍵由協定保留。
Warning
_meta 是你和用戶端應用程式之間的約定，不保證什麼會送到模型。要呈現什麼由 MCP 主機（host）決定。永遠不要在工具結果的任何部分放機密。
能力跟著處理函式走
Server 公告的方法族群，恰好就是你給了處理函式的那些。上面的 Bookshop 只傳了 on_list_tools 和 on_call_tool，其他什麼都沒有，所以連上它的用戶端會看到：
{"tools": {"listChanged": false}}
沒有 resources，沒有 prompts：背後沒有東西支撐它們。傳入 on_list_prompts，prompts 就會出現；傳入 on_completion，completions 就會出現。
MCPServer 不管你有沒有註冊，都一律公告工具、資源和提示詞，因為它的管理器永遠存在。在這一層，宣告就是那個建構子呼叫。
生命週期泛型
Server 對其生命週期 yield 出的型別是泛型的。註記一次，這個物件在每個出現的地方都有型別：
server.py
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from dataclasses import dataclass
from mcp.server import Server, ServerRequestContext
from mcp.types import (
CallToolRequestParams,
CallToolResult,
ListToolsResult,
PaginatedRequestParams,
TextContent,
Tool,
)
@dataclass
class Catalog:
books: list[str]
def search(self, query: str) -> list[str]:
return [title for title in self.books if query.lower() in title.lower()]
@asynccontextmanager
async def lifespan(server: Server[Catalog]) -> AsyncIterator[Catalog]:
yield Catalog(books=["Dune", "Dune Messiah", "Children of Dune"])
SEARCH_BOOKS = Tool(
name="search_books",
description="Search the catalog by title or author.",
input_schema={
"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"],
},
)
async def list_tools(ctx: ServerRequestContext[Catalog], params: PaginatedRequestParams | None) -> ListToolsResult:
return ListToolsResult(tools=[SEARCH_BOOKS])
async def call_tool(ctx: ServerRequestContext[Catalog], params: CallToolRequestParams) -> CallToolResult:
matches = ctx.lifespan_context.search((params.arguments or {})["query"])
text = f"Found {len(matches)} books: {', '.join(matches)}."
return CallToolResult(content=[TextContent(type="text", text=text)])
server = Server("Bookshop", lifespan=lifespan, on_list_tools=list_tools, on_call_tool=call_tool)
- 生命週期是一個
Callable[[Server[Catalog]], AbstractAsyncContextManager[Catalog]]；在async產生器上套@asynccontextmanager就正好得到這個。 - 它
yield出的東西會變成ctx.lifespan_context，而因為處理函式註記為ServerRequestContext[Catalog]，.search(...)可以自動完成，也能通過型別檢查。 - 伺服器啟動時進入一次，停止時離開一次。啟動、收尾，以及
MCPServer對同一個概念的版本，請見 生命週期。
沒有 lifespan= 的話，ctx.lifespan_context 是一個空的 dict。
自己的方法
建構子涵蓋 MCP 定義的方法。其他的一切由 add_request_handler 負責：
server.py
from pydantic import BaseModel
from mcp.server import Server, ServerRequestContext
from mcp.types import (
CallToolRequestParams,
CallToolResult,
ListToolsResult,
PaginatedRequestParams,
RequestParams,
TextContent,
Tool,
)
SEARCH_BOOKS = Tool(
name="search_books",
description="Search the catalog by title or author.",
input_schema={
"type": "object",
"properties": {"query": {"type": "string"}, "limit": {"type": "integer"}},
"required": ["query", "limit"],
},
)
async def list_tools(ctx: ServerRequestContext, params: PaginatedRequestParams | None) -> ListToolsResult:
return ListToolsResult(tools=[SEARCH_BOOKS])
async def call_tool(ctx: ServerRequestContext, params: CallToolRequestParams) -> CallToolResult:
args = params.arguments or {}
text = f"Found 3 books matching {args['query']!r} (showing up to {args['limit']})."
return CallToolResult(content=[TextContent(type="text", text=text)])
class ReindexParams(RequestParams):
full: bool = False
class ReindexResult(BaseModel):
indexed: int
async def reindex(ctx: ServerRequestContext, params: ReindexParams) -> ReindexResult:
return ReindexResult(indexed=3)
server = Server("Bookshop", on_list_tools=list_tools, on_call_tool=call_tool)
server.add_request_handler("bookshop/reindex", ReindexParams, reindex)
- 第一個引數是方法字串。通知有個孿生的
add_notification_handler。 params_type是傳入的params在處理函式執行之前用來驗證的模型，所以自訂方法確實享有工具沒有的驗證。繼承RequestParams，讓_meta欄位和其他方法一樣解析。- 處理函式回傳
BaseModel、dict或None。SDK 會把它序列化成 JSON-RPC 結果。
一個坦白的提醒：高階 Client 只有對應 MCP 定義方法的動詞，所以沒有 client.reindex()。廠商方法是給已經知道它存在的對端用的：你同時發佈的用戶端，或是你自己另一個講 JSON-RPC 的服務。
有一個方法你不能占用：
ValueError: 'initialize' is handled by the server runner and cannot be overridden;
use Server.middleware to observe or wrap initialization
交握屬於執行器。server/discover、ping，以及其他所有內建方法，都可以替換。
Tip
那則錯誤裡提到的 Server.middleware 會包住每一則傳入訊息，包括 initialize。如果想做的是觀察或改寫流量，而不是回應新方法，請從 中介軟體 開始。
其他處理函式
下面每一項都是一個你現在已經有詞彙可以理解的概念；每一項都有自己的頁面。
on_call_tool、on_get_prompt和on_read_resource可以回傳InputRequiredResult取代正常結果，暫停呼叫並向用戶端要求輸入；請見 多輪往返（multi-round-trip）請求。忠於這一層的風格，沒有任何東西會替你裝好：MCPServer預設會封裝requestState，在這裡你設定的request_state會一字不差地跨過線路，直到你用server.middleware.append(RequestStateBoundary(RequestStateSecurity(keys=[...]), default_audience=server.name))選擇加入：一行（兩個名稱都從mcp.server.request_state匯入）就能得到和MCPServer完全相同的封裝與驗證（保護requestState）。on_list_resources、on_read_resource、on_list_prompts、on_get_prompt、on_completion是其他基本元件的同一個(ctx, params) -> result形狀。on_subscriptions_listen負責 2026-07-28 的subscriptions/listen串流。傳入一個建構在SubscriptionBus之上的ListenHandler，並從其他處理函式把事件發佈到 bus；完整的組合方式請見 訂閱。server.streamable_http_app()回傳的 Starlette 應用程式和MCPServer的一樣；照 執行伺服器 部署其他 ASGI 應用程式的方式部署它。這一層沒有server.run(transport=...)：server.run(read_stream, write_stream, server.create_initialization_options())透過一對串流驅動一條連線，而這一行就是全部。
重點回顧
- 低階
Server以on_*建構子參數接收處理函式；每個處理函式都是async (ctx, params) -> result。 input_schemadict 自己寫，CallToolResult自己組。沒有任何東西會替你推導、包裝或驗證。- 處理函式裡的例外是
-32603協定錯誤。模型讀得到的工具錯誤，是你回傳的is_error=True的CallToolResult。 - 結果上的
_meta是給用戶端應用程式的，不是給模型的。 Server[T]對其生命週期 yield 出的東西是泛型的；ctx.lifespan_context是有型別的T。add_request_handler(method, params_type, handler)可以服務任何方法。initialize被保留。Server公告的能力，由你註冊了哪些處理函式推導而來。
Client(server) 對兩種伺服器一視同仁，因為它們就是同一個協定，這正是重點所在。再往下一層根本不是類別：是 中介軟體。