在使用 FastMCP 開發 MCP 服務器時經常會用到 @mcp.tool() 等裝飾器。雖然它們用起來很簡單，但當作黑匣子總讓人感覺"不得勁"。接下來我們將深入相關的源碼實現，別擔心，不會鉆沒有意義的“兔子洞”，你可以通過這篇文章了解到：

• 如何簡單啟動本地的 MCP Server 和 MCP Inspector
• 這些裝飾器具體做了什么
  • @mcp.tool()
  • @mcp.resource()
  • @mcp.prompt()

MCP 官方 Python SDK 地址：https://github.com/modelcontextprotocol/python-sdk。

代碼文件下載：server.py，debug_func_metadata.py，debug_message_validator.py

安裝庫

# 項目依賴已在 pyproject.toml 中配置，運行 uv sync 即可安裝
# 文章中重復的 uv add 是舊版本 pip install 的遺留（默認僅配置了 PyTorch 等基礎深度學習環境）
uv add mcp

server.py

下面是一個簡化的 server.py 示例：

from mcp.server.fastmcp import FastMCP# 初始化 FastMCP server
mcp = FastMCP(name="weather",#host="0.0.0.0",#port="8234"
)@mcp.tool()
def get_weather(city: str) -> str:"""獲取指定城市的天氣信息"""# 簡單模擬數據，實際應用中應該調用對應的APIweather_data = {"北京": "晴天，溫度 22°C","上海": "多云，溫度 25°C", "廣州": "小雨，溫度 28°C","深圳": "陰天，溫度 26°C"}return weather_data.get(city, f"{city} 的天氣數據暫不可用")@mcp.prompt()
def weather(city: str = "北京") -> list:"""提供天氣查詢的對話模板"""return [{"role": "user","content": f"請幫我查詢{city}的天氣情況，并提供詳細的天氣信息。"}]@mcp.resource("resource://cities")
def get_cities():"""返回支持查詢天氣的城市列表"""cities = ["北京", "上海", "廣州", "深圳"]return f"Cities: {', '.join(cities)}"@mcp.resource("resource://{city}/weather")
def get_city_weather(city: str) -> str:return f"Weather for {city}"if __name__ == "__main__":mcp.run(transport="stdio")

將其保存為 server 后，可以使用以下命令直接進行調試：

mcp dev server.py
# 如果克隆了倉庫，可以指定 Demos 文件夾下的路徑，比如：mcp dev Demos/mcp/server.py

mcp dev 會在運行MCP服務器的同時啟動 MCP Inspector：

MCP Inspector 界面配置如下圖左框：

配置項（Command + Arguments）實際對應于能夠運行服務器的命令，所以并不局限，有很多組合可以使用：

Command	Arguments
mcp	run server.py
python	server.py
uv	run server.py
uv	run mcp run server.py
uv	run --with mcp mcp run server.py
uv	run python server.py

最后三行命令實際只是 uv 對前兩行命令的封裝（uv 可以替代 pip/conda，目前已經被廣泛使用）。

連接成功后，你可以在 MCP Inspector 中看到注冊的 Resources、Prompts 和 Tools：

Resources	Prompts	Tools

什么是 FastMCP？

官方倉庫中對應的路徑為 src/mcp/server/fastmcp：

從 from mcp.server.fastmcp import FastMCP 開始，既然能夠直接 import FastMCP，那先查看 __init__.py：

"""FastMCP - 一個更人性化的 MCP 服務器接口。"""from importlib.metadata import versionfrom .server import Context, FastMCP
from .utilities.types import Image__version__ = version("mcp")
__all__ = ["FastMCP", "Context", "Image"]

可以看到 FastMCP 是從當前文件夾的 server.py 中導入的，所以接下來查看 server.py（省略部分初始化邏輯）：

class FastMCP:def __init__(self,name: str | None = None,instructions: str | None = None,auth_server_provider: OAuthAuthorizationServerProvider[Any, Any, Any] | None = None,token_verifier: TokenVerifier | None = None,event_store: EventStore | None = None,*,tools: list[Tool] | None = None,**settings: Any,):...self._tool_manager = ToolManager(tools=tools, warn_on_duplicate_tools=self.settings.warn_on_duplicate_tools)self._resource_manager = ResourceManager(warn_on_duplicate_resources=self.settings.warn_on_duplicate_resources)self._prompt_manager = PromptManager(warn_on_duplicate_prompts=self.settings.warn_on_duplicate_prompts)...

初始化（__init__）的代碼中有三個很眼熟的部分：

• _tool_manager：管理工具（Tools）
• _resource_manager：管理資源（Resources）
• _prompt_manager：管理提示詞（Prompts）

這三者分別對應于之后要介紹的裝飾器。

FastMCP 初始化的具體解析不會在本文進行，后續相關文章完結時此行會替換為索引鏈接。

裝飾器

什么是裝飾器？

裝飾器可以理解為一個接受函數作為參數，并返回一個新函數的函數。這可以讓我們在不修改原函數代碼的情況下添加通用的行為，通過一個簡單的例子來理解：

def decorator(func):"""一個簡單的裝飾器示例"""def wrapper(*args, **kwargs):print(f"調用函數 {func.__name__} 之前")result = func(*args, **kwargs)print(f"調用函數 {func.__name__} 之后\n")return resultreturn wrapper# 方式1：使用 @ 語法糖
@decorator
def say_hello(name):print(f"Hello, {name}!")
say_hello("Xiaoming")# 方式2：直接調用裝飾器函數
def say_hello(name):print(f"Hello, {name}!")
say_hello = decorator(say_hello)
say_hello("Xiaoming")

輸出：

調用函數 say_hello 之前
Hello, Xiaoming!
調用函數 say_hello 之后調用函數 say_hello 之前
Hello, Xiaoming!
調用函數 say_hello 之后

在 FastMCP 中，裝飾器的工作方式類似，但并不是簡單地 print，而是將函數注冊到對應的管理器中：

• @mcp.tool() - 將函數注冊為工具
• @mcp.resource() - 將函數注冊為資源
• @mcp.prompt() - 將函數注冊為提示詞模板

接下來會著重講解 tool() 裝飾器，resource() 和 prompt() 的處理邏輯基本是 tool() 的簡化版本，所以部分邏輯會帶過。

[!note]

使用 @mcp.* 的格式是因為初始化 mcp=FastMCP()，如果變量名從 mcp 改為了 server，即：server=FastMCP()，那么裝飾器就應該使用 @server.* 的格式。

@mcp.tool()

@mcp.tool() 裝飾器用于將 Python 函數自動注冊為當前 mcp 服務器中的工具。摘選之前的片段：

@mcp.tool()
def get_weather(city: str) -> str:"""獲取指定城市的天氣信息"""# 簡單模擬數據，實際應用中應該調用對應的APIweather_data = {"北京": "晴天，溫度 22°C","上海": "多云，溫度 25°C", "廣州": "小雨，溫度 28°C","深圳": "陰天，溫度 26°C"}return weather_data.get(city, f"{city} 的天氣數據暫不可用")

這段代碼實際上會：

1. 自動提取函數的參數類型信息，以及文檔字符串（下面的信息由之后的 debug_func_metadata 打印）。

   ============================================================
   🔍 開始解析函數: get_weather文檔字符串: 獲取指定城市的天氣信息
   ============================================================函數簽名分析:完整簽名: (city: str) -> str返回類型: <class 'str'>參數數量: 1
   參數名: city原始注解: <class 'str'>參數種類: POSITIONAL_OR_KEYWORD默認值: <class 'inspect._empty'>類型化注解: <class 'str'>字段信息: annotation=<class 'str'>, default=PydanticUndefined
   

2. 生成參數的 JSON Schema（函數名+Arguments）。

   🏗?  創建 Pydantic 模型:模型名稱: get_weatherArguments基類: <class 'mcp.server.fastmcp.utilities.func_metadata.ArgModelBase'>? 模型創建成功: <class '__main__.get_weatherArguments'>get_weatherArguments JSON Schema:
   {"properties": {"city": {"title": "City","type": "string"}},"required": ["city"],"title": "get_weatherArguments","type": "object"
   }
   

3. 將函數注冊為 MCP 工具（self._tools[tool.name] = tool）。

mcp.tool() 可以接受以下參數（此處參數解釋參考 tool 和 func_metadata）：

• name: 可選的工具名稱，默認為函數名

• title: 可選的工具標題（用于人類閱讀）

• description: 可選的工具功能描述，默認使用函數的文檔字符串

• annotations: 可選的 ToolAnnotations，提供額外的工具信息

• structured_output：控制工具輸出是結構化還是非結構化的

  • None: 基于函數的返回類型注解自動檢測

  • True: 無條件創建結構化工具（在返回類型注解允許的情況下）

    如果是結構化，會根據函數的返回類型注釋創建 Pydantic 模型。支持各種返回類型：

    • BaseModel 子類（直接使用）
    • 原始類型（str、int、float、bool、bytes、None）- 包裝在帶有 ‘result’ 字段的模型中
    • TypedDict - 轉換為具有相同字段的 Pydantic 模型
    • 數據類和其他帶注釋的類 - 轉換為 Pydantic 模型
    • 泛型類型（list、dict、Union 等）- 包裝在帶有 ‘result’ 字段的模型中

  • False: 無條件創建非結構化工具

[!note]

如果你只想了解如何使用 @mcp.tool()，可以跳過下面的源碼部分。

追溯源碼

class FastMCP:...def tool(self,name: str | None = None,title: str | None = None,description: str | None = None,annotations: ToolAnnotations | None = None,structured_output: bool | None = None,) -> Callable[[AnyFunction], AnyFunction]:"""用于注冊工具的裝飾器。工具可以通過添加 Context 類型注解的參數來可選地請求一個 Context 對象。Context 提供對 MCP 功能的訪問，包括日志記錄、進度報告和資源訪問。"""# 檢查裝飾器是否被正確使用（需要帶括號調用）if callable(name):raise TypeError("The @tool decorator was used incorrectly. Did you forget to call it? Use @tool() instead of @tool")def decorator(fn: AnyFunction) -> AnyFunction:self.add_tool(fn,name=name,title=title,description=description,annotations=annotations,structured_output=structured_output,)return fnreturn decorator

這里的 self.add_tool() 就是 mcp.add_tool()，所以也可以不使用裝飾器達到一樣的目的：

def get_weather():passmcp = FastMCP(name="weather")
mcp.add_tool(get_weather)  # 和 get_weather 使用@mcp.tool()效果一樣

這一行為最終調用的是 self._tool_manager.add_tool()，self._tool_manager 在__init__() 中對應的是 tools/tool_manager.py 中的 ToolManager 類：

class ToolManager:"""管理 FastMCP 工具."""def __init__(self,warn_on_duplicate_tools: bool = True,*,tools: list[Tool] | None = None,):self._tools: dict[str, Tool] = {}if tools is not None:for tool in tools:if warn_on_duplicate_tools and tool.name in self._tools:logger.warning(f"Tool already exists: {tool.name}")self._tools[tool.name] = toolself.warn_on_duplicate_tools = warn_on_duplicate_toolsdef add_tool(self,fn: Callable[..., Any],name: str | None = None,title: str | None = None,description: str | None = None,annotations: ToolAnnotations | None = None,structured_output: bool | None = None,) -> Tool:"""添加 tool 到 server。"""tool = Tool.from_function(fn,name=name,title=title,description=description,annotations=annotations,structured_output=structured_output,)existing = self._tools.get(tool.name)if existing:if self.warn_on_duplicate_tools:logger.warning(f"Tool already exists: {tool.name}")return existingself._tools[tool.name] = toolreturn tool

我們不需要關注 ToolManager 是怎么進行管理的，這不重要，重要的是裝飾器怎么處理我們自定義的函數。

整個 @mcp.tool() 裝飾器的工作按執行順序可以拆分為：

1. 裝飾器檢查調用方式是否正確（必須帶括號），然后將被裝飾的函數傳遞給 ToolManager.add_tool()。

   class FastMCP:...def tool(self, name, ...):if callable(name):raise TypeError("The @tool decorator was used incorrectly. Did you forget to call it? Use @tool() instead of @tool")def decorator(fn: AnyFunction) -> AnyFunction:self.add_tool(fn,  # 被裝飾的函數name=name,title=title,description=description,annotations=annotations,structured_output=structured_output,)return fnreturn decorator
   

2. Tool.from_function() 處理函數元數據。

   class ToolManager:...def add_tool(self, ...):tool = Tool.from_function(fn,  # 被裝飾的函數name=name,title=title,description=description,annotations=annotations,structured_output=structured_output,)...
   

   [!note]

   tool = Tool.from_function(...) 是當前最重要的處理部分，對應代碼位于 tools/base.py，其主要步驟如下（代碼按順序拼接等價于 Tool.from_function()，出于講解目的將其進行了拆分）：

   a. 解析函數簽名，提取參數信息。

   class Tool(BaseModel):...@classmethoddef from_function(cls,  # 這里的 cls 就是 Tool 類本身，不是實例fn: Callable[..., Any],name: str | None = None,title: str | None = None,description: str | None = None,context_kwarg: str | None = None,annotations: ToolAnnotations | None = None,structured_output: bool | None = None,) -> Tool:"""從函數創建工具."""from mcp.server.fastmcp.server import Contextfunc_name = name or fn.__name__# Lambda 函數必須提供 name 參數if func_name == "<lambda>":raise ValueError("You must provide a name for lambda functions")# 如果沒有傳入 description，則使用函數的文檔字符串func_doc = description or fn.__doc__ or ""is_async = _is_async_callable(fn)...
   

   b. 自動檢測 Context 參數， inspect.signature() 會遍歷函數的所有參數，檢查參數類型是否為 Context 的子類，如果是的話會記錄為 context_kwarg，這個參數會被傳入 func_metadata() 的 skip_names（可以跳過這一步的理解，等真正涉及到的時候再探究），不會出現在工具對應的 JSON Schema 中。

   class Tool(BaseModel):...@classmethoddef from_function(...):...# 自動檢測 Context 參數if context_kwarg is None:sig = inspect.signature(fn)for param_name, param in sig.parameters.items():# 跳過泛型類型if get_origin(param.annotation) is not None:continue# 檢查參數類型是否是 Context 的子類if issubclass(param.annotation, Context):context_kwarg = param_namebreak
   

   c. 生成參數的 JSON Schema：

   class Tool(BaseModel):...
   @classmethoddef from_function(...):...# 生成函數元數據，包括參數的 JSON Schemafunc_arg_metadata = func_metadata(fn,skip_names=[context_kwarg] if context_kwarg is not None else [],structured_output=structured_output,)# 從 Pydantic 模型生成 JSON Schemaparameters = func_arg_metadata.arg_model.model_json_schema(by_alias=True)
   

   func_metadata 相關源代碼位于 utilities/func_metadata.py，這里我們進行主體邏輯的抽取打印（完整 debug_func_metadata 函數見附錄）：

   def func(data,  # 無類型注解format: str = "json",  # 有注解+默認值count: Optional[int] = None,  # 復雜類型+默認值validate: bool = True  # 基礎類型+默認值
   ):  # 無返回類型注解"""展示各種注解情況"""return datadebug_func_metadata(func, skip_names="count")
   

   輸出：

   
   ============================================================
   🔍 開始解析函數: func文檔字符串: 展示各種注解情況
   ============================================================📋 函數簽名分析:完整簽名: (data, format: str = 'json', count: Optional[int] = None, validate: bool = True)返回類型: <class 'inspect._empty'>參數數量: 4🔧 參數處理詳情:跳過的參數名: ['c', 'o', 'u', 'n', 't'][1] 參數名: data原始注解: <class 'inspect._empty'>參數種類: POSITIONAL_OR_KEYWORD默認值: <class 'inspect._empty'>??  處理: 無類型注解，默認為 Any🔄 類型化注解: typing.Annotated[typing.Any, FieldInfo(annotation=NoneType, required=True), WithJsonSchema(json_schema={'title': 'data', 'type': 'string'}, mode=None)]? 字段信息: annotation=typing.Any, default=PydanticUndefined[2] 參數名: format原始注解: <class 'str'>參數種類: POSITIONAL_OR_KEYWORD默認值: json🔄 類型化注解: <class 'str'>? 字段信息: annotation=<class 'str'>, default=json[3] 參數名: count原始注解: typing.Optional[int]參數種類: POSITIONAL_OR_KEYWORD默認值: None??  跳過此參數[4] 參數名: validate原始注解: <class 'bool'>參數種類: POSITIONAL_OR_KEYWORD默認值: True🔄 類型化注解: <class 'bool'>? 字段信息: annotation=<class 'bool'>, default=True??  沖突處理: 參數名 'validate' 與 BaseModel 方法沖突-> 使用內部名稱: field_validate📊 參數處理總結:總參數數: 4處理參數數: 3模型字段: ['data', 'format', 'field_validate']🏗?  創建 Pydantic 模型:模型名稱: funcArguments基類: <class 'mcp.server.fastmcp.utilities.func_metadata.ArgModelBase'>? 模型創建成功: <class '__main__.funcArguments'>📄 funcArguments JSON Schema:
   {"properties": {"data": {"title": "data","type": "string"},"format": {"default": "json","title": "Format","type": "string"},"validate": {"default": true,"title": "Validate","type": "boolean"}},"required": ["data"],"title": "funcArguments","type": "object"
   }🎯 返回值處理:structured_output 參數: None返回注解: <class 'inspect._empty'>經過_get_typed_annotation處理后的類型: <class 'inspect._empty'>??  未創建輸出模型wrap_output: False? func_metadata 處理完成!最終結果: arg_model=<class '__main__.funcArguments'> output_schema=None output_model=None wrap_output=False
   ============================================================FuncMetadata(arg_model=<class '__main__.funcArguments'>, output_schema=None, output_model=None, wrap_output=False)
   

   d. 創建 Tool 實例（Tool.from_function 的 return cls(...)）。

   class Tool(BaseModel):...@classmethoddef from_function(...):return cls( # 使用 cls() 創建 Tool 實例，等價于 Tool()fn=fn,name=func_name,title=title,description=func_doc,parameters=parameters,fn_metadata=func_arg_metadata,is_async=is_async,context_kwarg=context_kwarg,annotations=annotations,)
   

3. 將 Tool 實例注冊到工具管理器中。

   class ToolManager:...def add_tool(...) -> Tool:"""添加 tool 到 server。"""tool = Tool.from_function(...)existing = self._tools.get(tool.name)if existing:if self.warn_on_duplicate_tools:logger.warning(f"Tool already exists: {tool.name}")return existingself._tools[tool.name] = toolreturn tool
   

@mcp.resource()

@mcp.resource() 裝飾器用于定義可供訪問的資源，需要注意的是：

• 必須提供一個資源 URI（如 @mcp.resource("resource://cities")）

• 資源可以是靜態的（每次調用返回相同內容）或動態的（根據參數填充內容）。

  • 靜態對應于 MCP Inspector 中的 Resources，動態對應于 Resources Templates，以下面兩個資源為例進行展示：

    from mcp.server.fastmcp import FastMCP# 初始化 FastMCP server
    mcp = FastMCP("cities")@mcp.resource("resource://cities")
    def get_cities():"""返回支持查詢天氣的城市列表"""cities = ["北京", "上海", "廣州", "深圳"]return f"Cities: {', '.join(cities)}"@mcp.resource("resource://{city}/weather")
    def get_city_weather(city: str) -> str:return f"Weather for {city}"if __name__ == "__main__":mcp.run(transport="stdio")
    

    此時 MCP Inspector 的 Resources 模塊顯示如下：

    

追溯源碼

查看 server.py 中的 resource 方法：

class FastMCP:...def resource(self,uri: str,*,name: str | None = None,title: str | None = None,description: str | None = None,mime_type: str | None = None,) -> Callable[[AnyFunction], AnyFunction]:"""用于將函數注冊為資源的裝飾器。當資源被讀取時，將調用被裝飾的函數來動態生成資源內容。函數可以返回：- str: 文本內容- bytes: 二進制內容  - 其他類型: 將自動轉換為 JSON 格式如果 URI 包含參數占位符（如 "resource://{param}"）或者函數本身有參數，該資源將被注冊為模板資源。參數：uri: 資源的 URI（如 "resource://my-resource" 或 "resource://{param}"）name: 可選的資源名稱title: 可選的資源標題（用于人類閱讀）description: 可選的資源描述mime_type: 可選的 MIME 類型使用示例：# 靜態資源@server.resource("resource://my-resource")def get_data() -> str:return "Hello, world!"# 參數化模板資源@server.resource("resource://{city}/weather")def get_weather(city: str) -> str:return f"Weather for {city}""""# 檢查裝飾器是否被正確使用（需要帶括號調用）if callable(uri):raise TypeError("The @resource decorator was used incorrectly. Did you forget to call it? Use @resource('uri') instead of @resource")def decorator(fn: AnyFunction) -> AnyFunction:# 通過 URI 中的 "{}" 和函數自身的參數來檢查是否是模版has_uri_params = "{" in uri and "}" in urihas_func_params = bool(inspect.signature(fn).parameters)if has_uri_params or has_func_params:# （有參數）提取 URI 參數和函數參數uri_params = set(re.findall(r"{(\w+)}", uri))func_params = set(inspect.signature(fn).parameters.keys())# 驗證 URI 參數和函數參數是否匹配if uri_params != func_params:raise ValueError(f"Mismatch between URI parameters {uri_params} and function parameters {func_params}")# 注冊為模板資源，調用 _resource_manager.add_template()self._resource_manager.add_template(fn=fn,uri_template=uri,name=name,title=title,description=description,mime_type=mime_type,)else:# （無參數）注冊為普通資源resource = FunctionResource.from_function(fn=fn,uri=uri,name=name,title=title,description=description,mime_type=mime_type,)self.add_resource(resource)  # 調用 self._resource_manager.add_resource(resource)return fnreturn decorator

ResourceManager 的實現位于 resources/resource_manager.py：

class ResourceManager:"""管理 FastMCP 資源。"""def __init__(self, warn_on_duplicate_resources: bool = True):self._resources: dict[str, Resource] = {}self._templates: dict[str, ResourceTemplate] = {}self.warn_on_duplicate_resources = warn_on_duplicate_resourcesdef add_resource(self, resource: Resource) -> Resource:"""向管理器中添加資源。參數：resource: 要添加的 Resource 實例返回：當前添加的資源。如果具有相同 URI 的資源已存在，則返回現有的資源。"""logger.debug("Adding resource",extra={"uri": resource.uri,"type": type(resource).__name__,"resource_name": resource.name,},)existing = self._resources.get(str(resource.uri))if existing:if self.warn_on_duplicate_resources:logger.warning(f"Resource already exists: {resource.uri}")return existingself._resources[str(resource.uri)] = resourcereturn resourcedef add_template(self,fn: Callable[..., Any],uri_template: str,name: str | None = None,title: str | None = None,description: str | None = None,mime_type: str | None = None,) -> ResourceTemplate:"""根據函數添加模版。"""template = ResourceTemplate.from_function(fn,uri_template=uri_template,name=name,title=title,description=description,mime_type=mime_type,)self._templates[template.uri_template] = templatereturn template...

對于靜態資源，add_resource 方法會直接將 FunctionResource 實例存儲在 _resources 字典中。對于動態資源，add_template 方法會創建 ResourceTemplate 實例并存儲在 _templates 字典中。

1. 靜態：FunctionResource 位于 resources/types.py：

   class FunctionResource(Resource):"""通過包裝函數來延遲加載數據的資源。函數只有在資源被讀取時才會被調用，允許對可能昂貴的數據進行延遲加載。這在列出資源時特別有用，因為函數不會被調用，直到資源被實際訪問。函數可以返回：- str 表示文本內容（默認）- bytes 表示二進制內容- 其他類型將被轉換為 JSON"""fn: Callable[[], Any] = Field(exclude=True)...@classmethoddef from_function(cls,fn: Callable[..., Any],uri: str,name: str | None = None,title: str | None = None,description: str | None = None,mime_type: str | None = None,) -> "FunctionResource":"""從函數創建 FunctionResource。"""func_name = name or fn.__name__if func_name == "<lambda>":raise ValueError("You must provide a name for lambda functions")# 確保參數被正確轉換fn = validate_call(fn)return cls(uri=AnyUrl(uri),name=func_name,title=title,description=description or fn.__doc__ or "",mime_type=mime_type or "text/plain",fn=fn,)
   

2. 動態：ResourceTemplate 位于 resources/templates.py：

   class ResourceTemplate(BaseModel):"""動態創建資源的模板。"""uri_template: str = Field(description="URI template with parameters (e.g. weather://{city}/current)")name: str = Field(description="Name of the resource")title: str | None = Field(description="Human-readable title of the resource", default=None)description: str | None = Field(description="Description of what the resource does")mime_type: str = Field(default="text/plain", description="MIME type of the resource content")fn: Callable[..., Any] = Field(exclude=True)parameters: dict[str, Any] = Field(description="JSON schema for function parameters")@classmethoddef from_function(cls,fn: Callable[..., Any],uri_template: str,name: str | None = None,title: str | None = None,description: str | None = None,mime_type: str | None = None,) -> ResourceTemplate:"""從函數創建模板。"""func_name = name or fn.__name__if func_name == "<lambda>":raise ValueError("You must provide a name for lambda functions")# 從 TypeAdapter 獲取 schema - 如果函數沒有正確的類型注解會失敗parameters = TypeAdapter(fn).json_schema()# 確保參數被正確轉換fn = validate_call(fn)return cls(uri_template=uri_template,name=func_name,title=title,description=description or fn.__doc__ or "",mime_type=mime_type or "text/plain",fn=fn,parameters=parameters,)
   

@mcp.prompt()

@mcp.prompt() 裝飾器用于定義提示詞模板，這部分的實現只是簡單維護了一個字典。

追溯源碼

查看 server.py 中的 prompt 方法：

class FastMCP:...def prompt(self,name: str | None = None,title: str | None = None,description: str | None = None,annotations: PromptAnnotations | None = None,) -> Callable[[AnyFunction], AnyFunction]:"""注冊提示詞的裝飾器。參數：name: 可選的提示詞名稱（默認使用函數名）title: 可選的提示詞人類可讀標題description: 可選的提示詞功能描述使用示例：@server.prompt()def analyze_table(table_name: str) -> list[Message]:schema = read_table_schema(table_name)return [{"role": "user","content": f"Analyze this schema:\n{schema}"}]@server.prompt()async def analyze_file(path: str) -> list[Message]:content = await read_file(path)return [{"role": "user","content": {"type": "resource","resource": {"uri": f"file://{path}","text": content}}}]"""# 同樣的驗證邏輯if callable(name):raise TypeError("The @prompt decorator was used incorrectly. Did you forget to call it? Use @prompt() instead of @prompt")def decorator(func: AnyFunction) -> AnyFunction:prompt = Prompt.from_function(func,name=name,title=title,description=description)self.add_prompt(prompt)  # 調用 self._prompt_manager.add_prompt(prompt)return funcreturn decorator

PromptManager 的實現位于 prompts/prompt_manager.py：

class PromptManager:"""管理 FastMCP 提示詞。"""def __init__(self, warn_on_duplicate_prompts: bool = True):self._prompts: dict[str, Prompt] = {}self.warn_on_duplicate_prompts = warn_on_duplicate_promptsdef add_prompt(self, prompt: Prompt) -> Prompt:"""添加提示詞到管理器。"""logger.debug(f"Adding prompt: {prompt.name}")existing = self._prompts.get(prompt.name)if existing:if self.warn_on_duplicate_prompts:logger.warning(f"Prompt already exists: {prompt.name}")return existingself._prompts[prompt.name] = promptreturn promptdef get_prompt(self, name: str) -> Prompt | None:"""根據名稱獲取提示詞。"""return self._prompts.get(name)def list_prompts(self) -> list[Prompt]:"""列出所有已注冊的提示詞。"""return list(self._prompts.values())

P.S. 文章跳過了 from_function 部分的源碼追溯（感興趣的同學可以點擊鏈接查看）。

[!note]

關于 @mcp.prompt() 的使用或許還需要多聊幾句，摘選之前的片段：

@mcp.prompt()
def weather(city: str = "北京") -> list:"""提供天氣查詢的對話模板"""return [{"role": "user","content": f"請幫我查詢{city}的天氣情況，并提供詳細的天氣信息。"}]

其實我們也可以這樣寫：

@mcp.prompt()
def weather(city: str = "北京") -> str:"""提供天氣查詢的對話模板"""return f"請幫我查詢{city}的天氣情況，并提供詳細的天氣信息。"

最終客戶端獲取的對象都是：

{"messages": [{"role": "user","content": {"type": "text","text": "請幫我查詢北京的天氣情況"}}]
}

簡單來說，如果被裝飾的函數直接返回字符串類型，就會被轉換為 UserMessage 對象（字典等類型的處理返回見附錄的 debug_message_validator.py 運行結果）。

那么，這個自動轉換的邏輯在哪實現呢？

當 MCP 客戶端請求提示詞時，FastMCP 會調用對應 Prompt 的 render() 方法：

class FastMCP:...async def get_prompt(self, name: str, arguments: dict[str, Any] | None = None) -> GetPromptResult:"""通過 name 和 arguments 獲取提示詞。"""try:prompt = self._prompt_manager.get_prompt(name)if not prompt:raise ValueError(f"Unknown prompt: {name}")# 調用 Prompt.render() 方法messages = await prompt.render(arguments)return GetPromptResult(description=prompt.description,messages=pydantic_core.to_jsonable_python(messages),)except Exception as e:logger.exception(f"Error getting prompt {name}")raise ValueError(str(e))

這個方法位于 prompts/base.py：

class Prompt(BaseModel):...async def render(self, arguments: dict[str, Any] | None = None) -> list[Message]:"""根據arguments渲染提示詞。"""# 驗證必需參數if self.arguments:required = {arg.name for arg in self.arguments if arg.required}provided = set(arguments or {})missing = required - providedif missing:raise ValueError(f"Missing required arguments: {missing}")try:# self.fn 就是被 mcp.prompt() 裝飾的函數# 這里是為了獲取 result（自定義函數執行后的返回值），并檢查是否為協程result = self.fn(**(arguments or {}))if inspect.iscoroutine(result):result = await result# 如果 result 不是列表或元組，轉換為列表if not isinstance(result, list | tuple):result = [result]# 轉換 result 為消息messages: list[Message] = []for msg in result:  # type: ignore[reportUnknownVariableType]try:if isinstance(msg, Message):# 如果是 Message 對象，直接使用messages.append(msg)elif isinstance(msg, dict):# 如果是字典，驗證并轉換為消息# message_validator = TypeAdapter[UserMessage | AssistantMessage](UserMessage | AssistantMessage)# Pydantic 的 TypeAdapter。用于驗證和轉換字典為 UserMessage 或 AssistantMessage 對象。# 當用戶返回字典格式的消息時，message_validator.validate_python(msg) 會根據字典中的 role 字段自動選擇合適的消息類型進行驗證和轉換。# https://docs.pydantic.dev/latest/api/type_adapter/?query=validate_pythonmessages.append(message_validator.validate_python(msg))elif isinstance(msg, str):# 如果是字符串，轉換為用戶消息content = TextContent(type="text", text=msg)messages.append(UserMessage(content=content))else:# 其他類型轉換為 JSON 字符串content = pydantic_core.to_json(msg, fallback=str, indent=2).decode()messages.append(Message(role="user", content=content))except Exception:raise ValueError(f"Could not convert prompt result to message: {msg}")return messagesexcept Exception as e:raise ValueError(f"Error rendering prompt {self.name}: {e}")

附錄

debug_func_metadata.py

官方源碼：utilities/func_metadata.py

調試文件下載：debug_func_metadata.py

#!/usr/bin/env python3
"""
func_metadata 調試工具，可以保存為 debug_func_metadata.py 執行
"""import inspect
import json
from typing import Any, Callable, Sequence, Optional, List, Dict
from pydantic import BaseModel, Field, create_model
from pydantic.fields import FieldInfo
from typing_extensions import Annotatedfrom mcp.server.fastmcp.utilities.func_metadata import (FuncMetadata, ArgModelBase,_get_typed_signature,_get_typed_annotation,_try_create_model_and_schema,InvalidSignature,PydanticUndefined,WithJsonSchema
)def print_json(data, title="JSON數據"):"""打印JSON數據"""try:print(f"\n📄 {title}:")print(json.dumps(data, indent=2, ensure_ascii=False))except Exception as e:print(f"? JSON打印失敗: {e}")print(f"原始數據類型: {type(data)}")print(f"原始數據: {data}")def debug_func_metadata(func: Callable[..., Any],skip_names: Sequence[str] = (),structured_output: bool | None = None,
) -> Any:"""調試版本的 func_metadata 實現，會用到一些 from_function 中的邏輯，比如：func.__name__，func.__doc__ ..."""print(f"\n{'='*60}")print(f"🔍 開始解析函數: {func.__name__}")print(f"   文檔字符串: {func.__doc__}")print(f"{'='*60}")try:# 從這里開始 func_metadata()# 步驟1: 獲取函數簽名sig = _get_typed_signature(func)params = sig.parametersprint(f"\n📋 函數簽名分析:")print(f"   完整簽名: {sig}")print(f"   返回類型: {sig.return_annotation}")print(f"   參數數量: {len(params)}")# 準備構建動態 Pydantic 模型的參數字典dynamic_pydantic_model_params: dict[str, Any] = {}globalns = getattr(func, "__globals__", {})print(f"\n🔧 參數處理詳情:")print(f"   跳過的參數名: {list(skip_names)}")# 步驟2: 遍歷每個參數processed_count = 0for idx, param in enumerate(params.values()):print(f"\n   [{idx+1}] 參數名: {param.name}")print(f"        原始注解: {param.annotation}")print(f"        參數種類: {param.kind}")print(f"        默認值: {param.default}")# 驗證參數名if param.name.startswith("_"):print(f"        ? 錯誤: 參數名不能以 '_' 開頭")raise InvalidSignature(f"{func.__name__} 的參數 {param.name} 不能以 '_' 開頭")if param.name in skip_names:print(f"        ??  跳過此參數")continueprocessed_count += 1annotation = param.annotation# 處理 `x: None` 或 `x: None = None` 的情況if annotation is None:print(f"        📝 處理: 類型為 None，添加默認值字段")annotation = Annotated[None,Field(default=param.default if param.default is not inspect.Parameter.empty else PydanticUndefined),]if annotation is inspect.Parameter.empty:print(f"        ??  處理: 無類型注解，默認為 Any")annotation = Annotated[Any,Field(),# 🤷 默認將無類型參數視為字符串WithJsonSchema({"title": param.name, "type": "string"}),]# 獲取類型化注解typed_annotation = _get_typed_annotation(annotation, globalns)print(f"        🔄 類型化注解: {typed_annotation}")# 創建字段信息field_info = FieldInfo.from_annotated_attribute(typed_annotation,param.default if param.default is not inspect.Parameter.empty else PydanticUndefined,)print(f"        ? 字段信息: annotation={field_info.annotation}, default={field_info.default}")# 處理參數名與 BaseModel 內置方法沖突的情況，這是必要的，因為 Pydantic 會因此發出警告# 例如：'dict' 或 'json' 等if hasattr(BaseModel, param.name) and callable(getattr(BaseModel, param.name)):print(f"        ??  沖突處理: 參數名 '{param.name}' 與 BaseModel 方法沖突")# 使用別名機制避免警告field_info.alias = param.namefield_info.validation_alias = param.namefield_info.serialization_alias = param.name# 內部使用帶前綴的參數名internal_name = f"field_{param.name}"dynamic_pydantic_model_params[internal_name] = (field_info.annotation, field_info)print(f"            -> 使用內部名稱: {internal_name}")else:dynamic_pydantic_model_params[param.name] = (field_info.annotation, field_info)print(f"\n📊 參數處理總結:")print(f"   總參數數: {len(params)}")print(f"   處理參數數: {processed_count}")print(f"   模型字段: {list(dynamic_pydantic_model_params.keys())}")# 步驟3: 動態創建一個 Pydantic 模型來表示函數參數arguments_model_name = f"{func.__name__}Arguments"print(f"\n🏗?  創建 Pydantic 模型:")print(f"   模型名稱: {arguments_model_name}")print(f"   基類: {ArgModelBase}")arguments_model = create_model(arguments_model_name,**dynamic_pydantic_model_params,__base__=ArgModelBase,)print(f"   ? 模型創建成功: {arguments_model}")# 生成并打印 JSON Schematry:# 這部分對應于 func_metadata() 之后的那行代碼，提前進行查看schema = arguments_model.model_json_schema(by_alias=True)print_json(schema, f"{arguments_model_name} JSON Schema")except Exception as e:print(f"? Schema 生成失敗: {e}")# 步驟4: 處理返回值（完全按照原版本邏輯）print(f"\n🎯 返回值處理:")print(f"   structured_output 參數: {structured_output}")print(f"   返回注解: {sig.return_annotation}")if structured_output is False:print(f"   🔚 明確不需要結構化輸出")result = FuncMetadata(arg_model=arguments_model)print(f"   ? 返回元數據: {result}")return result# 基于返回類型注釋設置結構化輸出支持if sig.return_annotation is inspect.Parameter.empty and structured_output is True:print(f"   ? 錯誤: 要求結構化輸出但無返回注解")raise InvalidSignature(f"函數 {func.__name__}: 結構化輸出需要返回注釋")output_info = FieldInfo.from_annotation(_get_typed_annotation(sig.return_annotation, globalns))annotation = output_info.annotationprint(f"      經過_get_typed_annotation處理后的類型: {annotation}")output_model, output_schema, wrap_output = _try_create_model_and_schema(annotation, func.__name__, output_info)if output_model:print(f"      ? 輸出模型創建成功: {output_model}")if output_schema:print_json(output_schema, "返回值 JSON Schema")else:print(f"      ??  未創建輸出模型")print(f"      wrap_output: {wrap_output}")# 模型創建失敗或產生警告 - 無結構化輸出if output_model is None and structured_output is True:print(f"      ? 結構化輸出失敗: 返回類型不可序列化")raise InvalidSignature(f"函數 {func.__name__}: 返回類型 {annotation} 不支持結構化輸出")# 創建最終結果result = FuncMetadata(arg_model=arguments_model,output_schema=output_schema,output_model=output_model,wrap_output=wrap_output,)print(f"\n? func_metadata 處理完成!")print(f"   最終結果: {result}")print(f"{'='*60}\n")return resultexcept Exception as e:print(f"? 處理過程中出錯: {e}")import tracebackprint(f"詳細錯誤信息:\n{traceback.format_exc()}")return Nonedef test():"""測試各種類型的函數"""# 混合注解print("\n\n📌 測試1: 混合類型注解")def func(data,  # 無類型注解format: str = "json",  # 有注解+默認值count: Optional[int] = None,  # 復雜類型+默認值validate: bool = True  # 基礎類型+默認值):  # 無返回類型注解"""展示各種注解情況"""return datadebug_func_metadata(func, skip_names="count")# 前綴參數測試print("\n\n📌 測試2: 前綴參數沖突")def prefix_func(_private: str, field_test: int) -> str:"""前綴參數"""return "test"debug_func_metadata(prefix_func)print("\n\n📌 測試3: 結構化輸出對比")def add(a: int, b: int) -> str:return a + bprint("📌 無結構化")debug_func_metadata(add, structured_output=False)print("\n\n📌 結構化")debug_func_metadata(add, structured_output=True)if __name__ == "__main__":test()

輸出：

📌 測試1: 混合類型注解============================================================
🔍 開始解析函數: func文檔字符串: 展示各種注解情況
============================================================📋 函數簽名分析:完整簽名: (data, format: str = 'json', count: Optional[int] = None, validate: bool = True)返回類型: <class 'inspect._empty'>參數數量: 4🔧 參數處理詳情:跳過的參數名: ['c', 'o', 'u', 'n', 't'][1] 參數名: data原始注解: <class 'inspect._empty'>參數種類: POSITIONAL_OR_KEYWORD默認值: <class 'inspect._empty'>??  處理: 無類型注解，默認為 Any🔄 類型化注解: typing.Annotated[typing.Any, FieldInfo(annotation=NoneType, required=True), WithJsonSchema(json_schema={'title': 'data', 'type': 'string'}, mode=None)]? 字段信息: annotation=typing.Any, default=PydanticUndefined[2] 參數名: format原始注解: <class 'str'>參數種類: POSITIONAL_OR_KEYWORD默認值: json🔄 類型化注解: <class 'str'>? 字段信息: annotation=<class 'str'>, default=json[3] 參數名: count原始注解: typing.Optional[int]參數種類: POSITIONAL_OR_KEYWORD默認值: None??  跳過此參數[4] 參數名: validate原始注解: <class 'bool'>參數種類: POSITIONAL_OR_KEYWORD默認值: True🔄 類型化注解: <class 'bool'>? 字段信息: annotation=<class 'bool'>, default=True??  沖突處理: 參數名 'validate' 與 BaseModel 方法沖突-> 使用內部名稱: field_validate📊 參數處理總結:總參數數: 4處理參數數: 3模型字段: ['data', 'format', 'field_validate']🏗?  創建 Pydantic 模型:模型名稱: funcArguments基類: <class 'mcp.server.fastmcp.utilities.func_metadata.ArgModelBase'>? 模型創建成功: <class '__main__.funcArguments'>📄 funcArguments JSON Schema:
{"properties": {"data": {"title": "data","type": "string"},"format": {"default": "json","title": "Format","type": "string"},"validate": {"default": true,"title": "Validate","type": "boolean"}},"required": ["data"],"title": "funcArguments","type": "object"
}🎯 返回值處理:structured_output 參數: None返回注解: <class 'inspect._empty'>經過_get_typed_annotation處理后的類型: <class 'inspect._empty'>??  未創建輸出模型wrap_output: False? func_metadata 處理完成!最終結果: arg_model=<class '__main__.funcArguments'> output_schema=None output_model=None wrap_output=False
============================================================📌 測試2: 前綴參數沖突============================================================
🔍 開始解析函數: prefix_func文檔字符串: 前綴參數
============================================================📋 函數簽名分析:完整簽名: (_private: str, field_test: int) -> str返回類型: <class 'str'>參數數量: 2🔧 參數處理詳情:跳過的參數名: [][1] 參數名: _private原始注解: <class 'str'>參數種類: POSITIONAL_OR_KEYWORD默認值: <class 'inspect._empty'>? 錯誤: 參數名不能以 '_' 開頭
? 處理過程中出錯: prefix_func 的參數 _private 不能以 '_' 開頭
詳細錯誤信息:
Traceback (most recent call last):File "/tmp/ipython-input-16-54014459.py", line 78, in debug_func_metadataraise InvalidSignature(f"{func.__name__} 的參數 {param.name} 不能以 '_' 開頭")
mcp.server.fastmcp.exceptions.InvalidSignature: prefix_func 的參數 _private 不能以 '_' 開頭📌 測試3: 結構化輸出對比
📌 無結構化============================================================
🔍 開始解析函數: add文檔字符串: None
============================================================📋 函數簽名分析:完整簽名: (a: int, b: int) -> str返回類型: <class 'str'>參數數量: 2🔧 參數處理詳情:跳過的參數名: [][1] 參數名: a原始注解: <class 'int'>參數種類: POSITIONAL_OR_KEYWORD默認值: <class 'inspect._empty'>🔄 類型化注解: <class 'int'>? 字段信息: annotation=<class 'int'>, default=PydanticUndefined[2] 參數名: b原始注解: <class 'int'>參數種類: POSITIONAL_OR_KEYWORD默認值: <class 'inspect._empty'>🔄 類型化注解: <class 'int'>? 字段信息: annotation=<class 'int'>, default=PydanticUndefined📊 參數處理總結:總參數數: 2處理參數數: 2模型字段: ['a', 'b']🏗?  創建 Pydantic 模型:模型名稱: addArguments基類: <class 'mcp.server.fastmcp.utilities.func_metadata.ArgModelBase'>? 模型創建成功: <class '__main__.addArguments'>📄 addArguments JSON Schema:
{"properties": {"a": {"title": "A","type": "integer"},"b": {"title": "B","type": "integer"}},"required": ["a","b"],"title": "addArguments","type": "object"
}🎯 返回值處理:structured_output 參數: False返回注解: <class 'str'>🔚 明確不需要結構化輸出? 返回元數據: arg_model=<class '__main__.addArguments'> output_schema=None output_model=None wrap_output=False📌 結構化============================================================
🔍 開始解析函數: add文檔字符串: None
============================================================📋 函數簽名分析:完整簽名: (a: int, b: int) -> str返回類型: <class 'str'>參數數量: 2🔧 參數處理詳情:跳過的參數名: [][1] 參數名: a原始注解: <class 'int'>參數種類: POSITIONAL_OR_KEYWORD默認值: <class 'inspect._empty'>🔄 類型化注解: <class 'int'>? 字段信息: annotation=<class 'int'>, default=PydanticUndefined[2] 參數名: b原始注解: <class 'int'>參數種類: POSITIONAL_OR_KEYWORD默認值: <class 'inspect._empty'>🔄 類型化注解: <class 'int'>? 字段信息: annotation=<class 'int'>, default=PydanticUndefined📊 參數處理總結:總參數數: 2處理參數數: 2模型字段: ['a', 'b']🏗?  創建 Pydantic 模型:模型名稱: addArguments基類: <class 'mcp.server.fastmcp.utilities.func_metadata.ArgModelBase'>? 模型創建成功: <class '__main__.addArguments'>📄 addArguments JSON Schema:
{"properties": {"a": {"title": "A","type": "integer"},"b": {"title": "B","type": "integer"}},"required": ["a","b"],"title": "addArguments","type": "object"
}🎯 返回值處理:structured_output 參數: True返回注解: <class 'str'>經過_get_typed_annotation處理后的類型: <class 'str'>? 輸出模型創建成功: <class 'mcp.server.fastmcp.utilities.func_metadata.addOutput'>📄 返回值 JSON Schema:
{"properties": {"result": {"title": "Result","type": "string"}},"required": ["result"],"title": "addOutput","type": "object"
}wrap_output: True? func_metadata 處理完成!最終結果: arg_model=<class '__main__.addArguments'> output_schema={'properties': {'result': {'title': 'Result', 'type': 'string'}}, 'required': ['result'], 'title': 'addOutput', 'type': 'object'} output_model=<class 'mcp.server.fastmcp.utilities.func_metadata.addOutput'> wrap_output=True
============================================================

debug_message_validator.py

官方源碼：prompts/base.py

調試文件下載：debug_message_validator.py

#!/usr/bin/env python3
"""
message_validator 調試工具，可以保存為 debug_message_validator.py 執行演示 FastMCP 中 message_validator.validate_python 的實際作用，
展示如何將字典轉換為 Message 對象，以及 Pydantic Union 類型的選擇行為。
"""from typing import Any, Literal
from pydantic import BaseModel, TypeAdapter
from mcp.types import ContentBlock, TextContentclass Message(BaseModel):"""基礎消息類 - MCP 協議中所有消息的基類"""role: Literal["user", "assistant"]content: ContentBlockdef __init__(self, content: str | ContentBlock, **kwargs: Any):# 如果內容是字符串，自動包裝為 TextContentif isinstance(content, str):content = TextContent(type="text", text=content)super().__init__(content=content, **kwargs)class UserMessage(Message):"""來自用戶的消息注意：role 字段允許 "user" 或 "assistant"，默認為 "user""""role: Literal["user", "assistant"] = "user"def __init__(self, content: str | ContentBlock, **kwargs: Any):super().__init__(content=content, **kwargs)class AssistantMessage(Message):"""來自助手的消息注意：role 字段允許 "user" 或 "assistant"，默認為 "assistant""""role: Literal["user", "assistant"] = "assistant"def __init__(self, content: str | ContentBlock, **kwargs: Any):super().__init__(content=content, **kwargs)# FastMCP 中的 message_validator 定義
# TypeAdapter 用于驗證和轉換數據為 Union 類型
message_validator = TypeAdapter[UserMessage | AssistantMessage](UserMessage | AssistantMessage)def demo_message_validator():"""調試版本的 message_validator 演示展示 FastMCP 中字典如何轉換為 Message 對象"""print(f"\n{'='*60}")print(f"🔍 開始調試 message_validator")print(f"   展示字典 → Message 對象的轉換過程")print(f"{'='*60}")# 步驟1: 分析類型定義print(f"\n📋 類型定義分析:")# 正確獲取 Pydantic 字段默認值user_role_field = UserMessage.model_fields.get('role')assistant_role_field = AssistantMessage.model_fields.get('role')user_default = user_role_field.default if user_role_field else "無字段"assistant_default = assistant_role_field.default if assistant_role_field else "無字段"print(f"   UserMessage 默認 role: {user_default}")print(f"   AssistantMessage 默認 role: {assistant_default}")print(f"   Union 類型順序: UserMessage | AssistantMessage")# 驗證實際行為print(f"\n🔧 實例化驗證:")user_instance = UserMessage(content="測試")assistant_instance = AssistantMessage(content="測試")print(f"   UserMessage() 實際 role: {user_instance.role}")print(f"   AssistantMessage() 實際 role: {assistant_instance.role}")# 準備測試用例test_cases = [{"name": "用戶消息字典","data": {"role": "user","content": "簡單的文本消息"}},{"name": "助手消息字典","data": {"role": "assistant","content": "我是助手的回復"}}]print(f"\n🔧 轉換測試詳情:")print(f"   測試數量: {len(test_cases)}")# 步驟2: 執行轉換測試for idx, test_case in enumerate(test_cases, 1):print(f"\n   [{idx}] 測試名稱: {test_case['name']}")print(f"        輸入數據: {test_case['data']}")print(f"        字段分析:")print(f"            role = '{test_case['data']['role']}'")print(f"            content = '{test_case['data']['content']}'")try:# 調用 message_validator.validate_pythonresult = message_validator.validate_python(test_case['data'])print(f"        ? 轉換成功!")print(f"        🔄 轉換結果:")print(f"            類型: {type(result).__name__}")print(f"            角色: {result.role}")print(f"            內容類型: {type(result.content).__name__}")if hasattr(result.content, 'text'):print(f"            內容文本: {result.content.text}")# 分析異常情況if test_case['data']['role'] == 'assistant' and isinstance(result, UserMessage):print(f"        ??  異常發現: role='assistant' 但返回了 UserMessage")print(f"            -> 原因: Pydantic Union 按順序驗證")print(f"            -> UserMessage 也接受 role='assistant'")print(f"            -> 第一個成功驗證的類型被選擇")except Exception as e:print(f"        ? 轉換失敗:")print(f"            錯誤類型: {type(e).__name__}")print(f"            錯誤信息: {e}")# 步驟3: 測試錯誤處理print(f"\n📊 錯誤處理測試:")print(f"   測試 message_validator 的錯誤處理能力")# 準備錯誤測試用例error_cases = [{"name": "缺少 role 字段","data": {"content": "沒有角色信息"},"expected": "應該使用默認角色或報錯"},{"name": "錯誤的 role 值","data": {"role": "system",  # 不支持的角色"content": "系統消息"},"expected": "應該驗證失敗"},{"name": "缺少 content 字段","data": {"role": "user"},"expected": "必需字段缺失"}]for idx, test_case in enumerate(error_cases, 1):print(f"\n   [{idx}] 錯誤場景: {test_case['name']}")print(f"        輸入數據: {test_case['data']}")print(f"        預期行為: {test_case['expected']}")try:result = message_validator.validate_python(test_case['data'])print(f"        ? 意外成功!")print(f"            結果: {result}")print(f"            類型: {type(result).__name__}")except Exception as e:print(f"        ? 預期的錯誤:")print(f"            錯誤類型: {type(e).__name__}")print(f"            錯誤信息: {str(e)}")def demo_prompt_render_simulation():"""模擬 Prompt.render() 中的消息轉換過程展示用戶函數返回值如何被處理成標準 MCP 消息"""print(f"\n{'='*60}")print(f"🔄 模擬 Prompt.render() 消息轉換")print(f"   展示用戶函數返回值 → 標準 MCP 消息的過程")print(f"{'='*60}")# 準備用戶函數可能返回的各種類型user_returns = ["簡單字符串",{"role": "user","content": "字典格式的用戶消息"},{"role": "assistant","content": "字典格式的助手消息"},UserMessage(content="直接的 UserMessage 對象"),AssistantMessage(content="直接的 AssistantMessage 對象"),["多個", "字符串"],["混合類型",{"role": "user", "content": "字典消息"},AssistantMessage(content="對象消息")]]print(f"\n📋 模擬場景分析:")print(f"   返回類型數量: {len(user_returns)}")print(f"   覆蓋場景: 字符串、字典、對象、列表、混合類型")def simulate_render_conversion(result):"""模擬 render() 方法中的轉換邏輯按照 FastMCP 的實際處理順序進行轉換"""print(f"        🔄 開始轉換處理:")print(f"            原始類型: {type(result).__name__}")# 步驟1: 規范化為列表if not isinstance(result, list | tuple):result = [result]print(f"            -> 單項轉為列表: [1項]")else:print(f"            -> 已是列表: [{len(result)}項]")# 步驟2: 逐項轉換為消息messages = []for idx, msg in enumerate(result, 1):print(f"            項目{idx}: {type(msg).__name__}")try:if isinstance(msg, Message):# Message 對象直接使用messages.append(msg)print(f"                ? 直接使用: {type(msg).__name__}({msg.role})")print(f"                   內容: {str(msg.content)}")elif isinstance(msg, dict):# 字典通過 message_validator 轉換converted = message_validator.validate_python(msg)messages.append(converted)print(f"                🔄 字典轉換: {msg}")print(f"                   結果: {type(converted).__name__}({converted.role})")elif isinstance(msg, str):# 字符串包裝為 UserMessagecontent = TextContent(type="text", text=msg)user_msg = UserMessage(content=content)messages.append(user_msg)print(f"                📝 字符串轉換: '{msg}'")print(f"                   結果: UserMessage(user)")else:# 其他類型序列化為 JSONimport jsoncontent_str = json.dumps(msg, ensure_ascii=False, indent=2)user_msg = UserMessage(content=content_str)messages.append(user_msg)print(f"                📦 JSON轉換: {type(msg).__name__}")print(f"                   結果: UserMessage(user)")except Exception as e:print(f"                ? 轉換失敗: {str(e)}")return messages# 步驟3: 執行場景測試print(f"\n🔧 場景測試詳情:")for idx, user_return in enumerate(user_returns, 1):print(f"\n   [{idx}] 場景名稱: 用戶返回 {type(user_return).__name__}")print(f"        原始數據: {str(user_return)}")messages = simulate_render_conversion(user_return)print(f"        📊 轉換總結:")print(f"            生成消息數: {len(messages)}")for msg_idx, msg in enumerate(messages, 1):print(f"            消息{msg_idx}: {type(msg).__name__}({msg.role})")print(f"                     內容: {str(msg.content)}")def debug_pydantic_union_behavior():"""深入分析 Pydantic Union 類型選擇行為解釋為什么 role='assistant' 時返回 UserMessage 而不是 AssistantMessage"""print(f"\n{'='*60}")print(f"🔍 深入分析 Pydantic Union 類型選擇")print(f"   解釋 Union 類型的驗證順序和選擇邏輯")print(f"{'='*60}")# 準備測試數據test_data = {"role": "assistant","content": "助手消息"}print(f"\n📋 測試數據分析:")print(f"   輸入數據: {test_data}")print(f"   預期類型: AssistantMessage（因為 role='assistant'）")print(f"\n🔧 驗證步驟:")# 步驟1: 直接構造 UserMessageprint(f"\n   [1] 直接構造 UserMessage:")try:user_msg = UserMessage(**test_data)print(f"        ? 構造成功!")print(f"            結果類型: {type(user_msg).__name__}")print(f"            角色字段: {user_msg.role}")print(f"            ??  說明: UserMessage 接受 role='assistant'")except Exception as e:print(f"        ? 構造失敗: {e}")# 步驟2: 直接構造 AssistantMessageprint(f"\n   [2] 直接構造 AssistantMessage:")try:assistant_msg = AssistantMessage(**test_data)print(f"        ? 構造成功!")print(f"            結果類型: {type(assistant_msg).__name__}")print(f"            角色字段: {assistant_msg.role}")print(f"            ??  說明: AssistantMessage 也接受 role='assistant'")except Exception as e:print(f"        ? 構造失敗: {e}")# 步驟3: TypeAdapter 選擇print(f"\n   [3] TypeAdapter Union 選擇:")try:adapter_result = message_validator.validate_python(test_data)print(f"        📊 最終結果:")print(f"            選擇類型: {type(adapter_result).__name__}")print(f"            角色字段: {adapter_result.role}")except Exception as e:print(f"        ? TypeAdapter 轉換失敗: {e}")print(f"\n? 結論總結:")print(f"   📝 核心原理: Pydantic Union 按順序驗證")print(f"       1. Union[UserMessage, AssistantMessage] 先驗證 UserMessage")print(f"       2. UserMessage.role 允許 'user' | 'assistant'")print(f"       3. role='assistant' 通過 UserMessage 驗證")print(f"       4. 驗證成功，返回 UserMessage 實例")print(f"       5. 不再嘗試 AssistantMessage")print(f"   🎯 實際影響:")print(f"       - role='assistant' 總是返回 UserMessage")print(f"       - 只有明確指定類型才能獲得 AssistantMessage")print(f"       - 或許是 bug，但本來二者的定義就一樣，只是類名不同，不再繼續深究")def test():"""測試各種類型的 message_validator 行為"""print("\n\n📌 測試1: 基礎轉換行為")demo_message_validator()print("\n\n📌 測試2: Prompt.render() 模擬")demo_prompt_render_simulation()print("\n\n📌 測試3: Pydantic Union 選擇分析")debug_pydantic_union_behavior()if __name__ == "__main__":test()

輸出：

📌 測試1: 基礎轉換行為============================================================
🔍 開始調試 message_validator展示字典 → Message 對象的轉換過程
============================================================📋 類型定義分析:UserMessage 默認 role: userAssistantMessage 默認 role: assistantUnion 類型順序: UserMessage | AssistantMessage🔧 實例化驗證:UserMessage() 實際 role: userAssistantMessage() 實際 role: assistant🔧 轉換測試詳情:測試數量: 2[1] 測試名稱: 用戶消息字典輸入數據: {'role': 'user', 'content': '簡單的文本消息'}字段分析:role = 'user'content = '簡單的文本消息'? 轉換成功!🔄 轉換結果:類型: UserMessage角色: user內容類型: TextContent內容文本: 簡單的文本消息[2] 測試名稱: 助手消息字典輸入數據: {'role': 'assistant', 'content': '我是助手的回復'}字段分析:role = 'assistant'content = '我是助手的回復'? 轉換成功!🔄 轉換結果:類型: UserMessage角色: assistant內容類型: TextContent內容文本: 我是助手的回復??  異常發現: role='assistant' 但返回了 UserMessage-> 原因: Pydantic Union 按順序驗證-> UserMessage 也接受 role='assistant'-> 第一個成功驗證的類型被選擇📊 錯誤處理測試:測試 message_validator 的錯誤處理能力[1] 錯誤場景: 缺少 role 字段輸入數據: {'content': '沒有角色信息'}預期行為: 應該使用默認角色或報錯? 意外成功!結果: role='user' content=TextContent(type='text', text='沒有角色信息', annotations=None, meta=None)類型: UserMessage[2] 錯誤場景: 錯誤的 role 值輸入數據: {'role': 'system', 'content': '系統消息'}預期行為: 應該驗證失敗? 預期的錯誤:錯誤類型: ValidationError錯誤信息: 2 validation errors for union[UserMessage,AssistantMessage]
UserMessage.roleInput should be 'user' or 'assistant' [type=literal_error, input_value='system', input_type=str]For further information visit https://errors.pydantic.dev/2.9/v/literal_error
AssistantMessage.roleInput should be 'user' or 'assistant' [type=literal_error, input_value='system', input_type=str]For further information visit https://errors.pydantic.dev/2.9/v/literal_error[3] 錯誤場景: 缺少 content 字段輸入數據: {'role': 'user'}預期行為: 必需字段缺失? 預期的錯誤:錯誤類型: TypeError錯誤信息: UserMessage.__init__() missing 1 required positional argument: 'content'📌 測試2: Prompt.render() 模擬============================================================
🔄 模擬 Prompt.render() 消息轉換展示用戶函數返回值 → 標準 MCP 消息的過程
============================================================📋 模擬場景分析:返回類型數量: 7覆蓋場景: 字符串、字典、對象、列表、混合類型🔧 場景測試詳情:[1] 場景名稱: 用戶返回 str原始數據: 簡單字符串🔄 開始轉換處理:原始類型: str-> 單項轉為列表: [1項]項目1: str📝 字符串轉換: '簡單字符串'結果: UserMessage(user)📊 轉換總結:生成消息數: 1消息1: UserMessage(user)內容: type='text' text='簡單字符串' annotations=None meta=None[2] 場景名稱: 用戶返回 dict原始數據: {'role': 'user', 'content': '字典格式的用戶消息'}🔄 開始轉換處理:原始類型: dict-> 單項轉為列表: [1項]項目1: dict🔄 字典轉換: {'role': 'user', 'content': '字典格式的用戶消息'}結果: UserMessage(user)📊 轉換總結:生成消息數: 1消息1: UserMessage(user)內容: type='text' text='字典格式的用戶消息' annotations=None meta=None[3] 場景名稱: 用戶返回 dict原始數據: {'role': 'assistant', 'content': '字典格式的助手消息'}🔄 開始轉換處理:原始類型: dict-> 單項轉為列表: [1項]項目1: dict🔄 字典轉換: {'role': 'assistant', 'content': '字典格式的助手消息'}結果: UserMessage(assistant)📊 轉換總結:生成消息數: 1消息1: UserMessage(assistant)內容: type='text' text='字典格式的助手消息' annotations=None meta=None[4] 場景名稱: 用戶返回 UserMessage原始數據: role='user' content=TextContent(type='text', text='直接的 UserMessage 對象', annotations=None, meta=None)🔄 開始轉換處理:原始類型: UserMessage-> 單項轉為列表: [1項]項目1: UserMessage? 直接使用: UserMessage(user)內容: type='text' text='直接的 UserMessage 對象' annotations=None meta=None📊 轉換總結:生成消息數: 1消息1: UserMessage(user)內容: type='text' text='直接的 UserMessage 對象' annotations=None meta=None[5] 場景名稱: 用戶返回 AssistantMessage原始數據: role='assistant' content=TextContent(type='text', text='直接的 AssistantMessage 對象', annotations=None, meta=None)🔄 開始轉換處理:原始類型: AssistantMessage-> 單項轉為列表: [1項]項目1: AssistantMessage? 直接使用: AssistantMessage(assistant)內容: type='text' text='直接的 AssistantMessage 對象' annotations=None meta=None📊 轉換總結:生成消息數: 1消息1: AssistantMessage(assistant)內容: type='text' text='直接的 AssistantMessage 對象' annotations=None meta=None[6] 場景名稱: 用戶返回 list原始數據: ['多個', '字符串']🔄 開始轉換處理:原始類型: list-> 已是列表: [2項]項目1: str📝 字符串轉換: '多個'結果: UserMessage(user)項目2: str📝 字符串轉換: '字符串'結果: UserMessage(user)📊 轉換總結:生成消息數: 2消息1: UserMessage(user)內容: type='text' text='多個' annotations=None meta=None消息2: UserMessage(user)內容: type='text' text='字符串' annotations=None meta=None[7] 場景名稱: 用戶返回 list原始數據: ['混合類型', {'role': 'user', 'content': '字典消息'}, AssistantMessage(role='assistant', content=TextContent(type='text', text='對象消息', annotations=None, meta=None))]🔄 開始轉換處理:原始類型: list-> 已是列表: [3項]項目1: str📝 字符串轉換: '混合類型'結果: UserMessage(user)項目2: dict🔄 字典轉換: {'role': 'user', 'content': '字典消息'}結果: UserMessage(user)項目3: AssistantMessage? 直接使用: AssistantMessage(assistant)內容: type='text' text='對象消息' annotations=None meta=None📊 轉換總結:生成消息數: 3消息1: UserMessage(user)內容: type='text' text='混合類型' annotations=None meta=None消息2: UserMessage(user)內容: type='text' text='字典消息' annotations=None meta=None消息3: AssistantMessage(assistant)內容: type='text' text='對象消息' annotations=None meta=None📌 測試3: Pydantic Union 選擇分析============================================================
🔍 深入分析 Pydantic Union 類型選擇解釋 Union 類型的驗證順序和選擇邏輯
============================================================📋 測試數據分析:輸入數據: {'role': 'assistant', 'content': '助手消息'}預期類型: AssistantMessage（因為 role='assistant'）🔧 驗證步驟:[1] 直接構造 UserMessage:? 構造成功!結果類型: UserMessage角色字段: assistant??  說明: UserMessage 接受 role='assistant'[2] 直接構造 AssistantMessage:? 構造成功!結果類型: AssistantMessage角色字段: assistant??  說明: AssistantMessage 也接受 role='assistant'[3] TypeAdapter Union 選擇:📊 最終結果:選擇類型: UserMessage角色字段: assistant? 結論總結:📝 核心原理: Pydantic Union 按順序驗證1. Union[UserMessage, AssistantMessage] 先驗證 UserMessage2. UserMessage.role 允許 'user' | 'assistant'3. role='assistant' 通過 UserMessage 驗證4. 驗證成功，返回 UserMessage 實例5. 不再嘗試 AssistantMessage🎯 實際影響:- role='assistant' 總是返回 UserMessage- 只有明確指定類型才能獲得 AssistantMessage- 或許是 bug，但本來二者的定義就一樣，只是類名不同，不再繼續深究