1. 结构化输出 #

本节将介绍如何实现和使用结构化输出。结构化输出是指不再仅仅通过自然语言文本与工具交互，而是能够让工具以及模型输出明确的数据结构（如 JSON、对象、表格等），从而更精确地表达调用结果。
、tools_structured_client.py 演示了如何获取并打印结构化内容 (structured_content 字段)，区别于仅仅处理文本返回。这一设计极大提升了工具调用和模型交互的可用性和下游自动化能力。

你将学会如何读取结构化数据、与文本内容结合打印，并理解何时使用结构化内容优先于普通文本输出。

npx @modelcontextprotocol/inspector uv --directory D:/aprepare/mcp-starter run tools_structured_server.py

2. tools_structured_client.py #

tools_structured_client.py

# 导入 os 库，用于文件路径操作
import os

# 从 mcp_lite 包导入客户端会话类、标准输入输出服务器参数类和相关类型定义
from mcp_lite import ClientSession, StdioServerParameters, types

# 从 mcp_lite.client.stdio 模块导入 stdio_client 工厂函数
from mcp_lite.client.stdio import stdio_client

# 定义一个用于打印工具调用结果的辅助函数
def print_call_result(tag: str, result: types.CallToolResult) -> None:
    # readable_parts 用于保存解析后的文本内容
    readable_parts: list[str] = []
    # 遍历 content 列表中的每个块
    for block in result.content:
        # 如果块是 TextContent 类型，提取文本并加入 readable_parts
        if isinstance(block, types.TextContent):
            readable_parts.append(block.text)
        # 如果块是字典并且类型为 "text"，提取并加入 readable_parts
        elif isinstance(block, dict) and block.get("type") == "text":
            readable_parts.append(block.get("text", ""))
    # 将所有 readable_parts 拼接成一个字符串，若为空则设置默认文本
    printable_text = " | ".join(readable_parts) if readable_parts else "<no text content>"
    # 获取结构化内容字段 structured_content
    structured = getattr(result, "structured_content", None)
    # 打印文本内容
    print(f"[{tag}] text=", printable_text)
    # 打印结构化内容
    print(f"[{tag}] structured=", structured)

# 主函数，客户端入口
def main() -> None:
    # 获取当前文件所在目录
    base_dir = os.path.dirname(os.path.abspath(__file__))
    # 拼接 server 脚本路径
    server_path = os.path.join(base_dir, "tools_structured_server.py")

    # 构造用于启动服务器的参数
    server_params = StdioServerParameters(
        command="python",        # 指定启动命令为 python
        args=[server_path],      # 启动参数为 server 脚本路径
        env={},                  # 使用空环境变量字典
    )

    # 使用 stdio_client 启动服务器（with 自动管理资源）
    with stdio_client(server_params) as (read, write):
        # 创建客户端会话实例
        session = ClientSession(read, write)
        # 执行初始化，完成握手
        session.initialize()

        # 获取工具列表并打印其名称
        tools = session.list_tools()
        print("[Tools]", [t.name for t in tools.tools])

        # 调用 get_weather 工具并打印结果
        res_weather = session.call_tool("get_weather", {"city": "Hangzhou"})
        print_call_result("get_weather", res_weather)

        # 调用 get_location 工具并打印结果
        res_location = session.call_tool("get_location", {"address": "天安门"})
        print_call_result("get_location", res_location)

        # 调用 get_statistics 工具并打印结果
        res_stats = session.call_tool("get_statistics", {"data_type": "scores"})
        print_call_result("get_statistics", res_stats)

        # 调用 get_user 工具并打印结果
        res_user = session.call_tool("get_user", {"user_id": "u001"})
        print_call_result("get_user", res_user)

        # 调用 get_temperature 工具并打印结果
        res_temp = session.call_tool("get_temperature", {"city": "Beijing"})
        print_call_result("get_temperature", res_temp)

        # 调用 list_cities 工具并打印结果
        res_cities = session.call_tool("list_cities", {})
        print_call_result("list_cities", res_cities)

        # 调用 get_config 工具并打印结果
        res_config = session.call_tool("get_config", {})
        print_call_result("get_config", res_config)

        # 调用 unstructured_message 工具并打印结果
        res_unstruct = session.call_tool("unstructured_message", {})
        print_call_result("unstructured_message", res_unstruct)

# 判断是否作为主程序入口执行
if __name__ == "__main__":
    main()

官方代码

# 导入 os 库，用于文件路径操作
import os
import asyncio
# 从 mcp_lite 包导入客户端会话类、标准输入输出服务器参数类和相关类型定义
from mcp import ClientSession, StdioServerParameters, types

# 从 mcp_lite.client.stdio 模块导入 stdio_client 工厂函数
from mcp.client.stdio import stdio_client

# 定义一个用于打印工具调用结果的辅助函数
def print_call_result(tag: str, result: types.CallToolResult) -> None:
    # readable_parts 用于保存解析后的文本内容
    readable_parts: list[str] = []
    # 遍历 content 列表中的每个块
    for block in result.content:
        # 如果块是 TextContent 类型，提取文本并加入 readable_parts
        if isinstance(block, types.TextContent):
            readable_parts.append(block.text)
        # 如果块是字典并且类型为 "text"，提取并加入 readable_parts
        elif isinstance(block, dict) and block.get("type") == "text":
            readable_parts.append(block.get("text", ""))
    # 将所有 readable_parts 拼接成一个字符串，若为空则设置默认文本
    printable_text = " | ".join(readable_parts) if readable_parts else "<no text content>"
    # 获取结构化内容字段 structured_content
    structured = getattr(result, "structured_content", None)
    # 打印文本内容
    print(f"[{tag}] text=", printable_text)
    # 打印结构化内容
    print(f"[{tag}] structured=", structured)

# 主函数，客户端入口
async def main() -> None:
    # 获取当前文件所在目录
    base_dir = os.path.dirname(os.path.abspath(__file__))
    # 拼接 server 脚本路径
    server_path = os.path.join(base_dir, "tools_structured_server.py")

    # 构造用于启动服务器的参数
    server_params = StdioServerParameters(
        command="python",        # 指定启动命令为 python
        args=[server_path],      # 启动参数为 server 脚本路径
        env={},                  # 使用空环境变量字典
    )

    # 使用 stdio_client 启动服务器（with 自动管理资源）
    async with stdio_client(server_params) as (read, write):
        # 创建客户端会话实例
        async with ClientSession(read, write) as session:
            # 执行初始化，完成握手
            await session.initialize()

            # 获取工具列表并打印其名称
            tools = await session.list_tools()
            print("[Tools]", [t.name for t in tools.tools])

            # 调用 get_weather 工具并打印结果
            res_weather = await session.call_tool("get_weather", {"city": "Hangzhou"})
            print_call_result("get_weather", res_weather)

            # 调用 get_location 工具并打印结果
            res_location = await session.call_tool("get_location", {"address": "天安门"})
            print_call_result("get_location", res_location)

            # 调用 get_statistics 工具并打印结果
            res_stats = await session.call_tool("get_statistics", {"data_type": "scores"})
            print_call_result("get_statistics", res_stats)

            # 调用 get_user 工具并打印结果
            res_user = await session.call_tool("get_user", {"user_id": "u001"})
            print_call_result("get_user", res_user)

            # 调用 get_temperature 工具并打印结果
            res_temp = await session.call_tool("get_temperature", {"city": "Beijing"})
            print_call_result("get_temperature", res_temp)

            # 调用 list_cities 工具并打印结果
            res_cities = await session.call_tool("list_cities", {})
            print_call_result("list_cities", res_cities)

            # 调用 get_config 工具并打印结果
            res_config = await session.call_tool("get_config", {})
            print_call_result("get_config", res_config)

            # 调用 unstructured_message 工具并打印结果
            res_unstruct = await session.call_tool("unstructured_message", {})
            print_call_result("unstructured_message", res_unstruct)

# 判断是否作为主程序入口执行
if __name__ == "__main__":
    asyncio.run(main())

3. tools_structured_server.py #

tools_structured_server.py

# 工具结构化服务器示例，演示多种结构化输出方式与如何禁用结构化输出

# 导入 sys 模块，用于设置标准流编码
import sys
# 导入 TypedDict 用于类型提示
from typing import TypedDict

# 导入 Pydantic 的 BaseModel 和 Field，用于定义结构化数据模型
from pydantic import BaseModel, Field

# 从 mcp_lite 导入 FastMCP 服务器框架
from mcp_lite.server.fastmcp import FastMCP

# 创建FastMCP服务器实例，设置服务名为"Tools Structured Output"
mcp = FastMCP(name="Tools Structured Output")

# 一、定义结构化天气数据模型，继承自Pydantic BaseModel
class WeatherData(BaseModel):
    # 温度字段，浮点型，带描述
    temperature: float = Field(description="摄氏温度")
    # 湿度字段，浮点型，带描述
    humidity: float = Field(description="湿度百分比")
    # 天气状况，字符串
    condition: str = Field(description="天气状况")

# 注册get_weather为工具，根据城市返回结构化天气数据
@mcp.tool()
def get_weather(city: str) -> WeatherData:
    # 构造结构化天气数据返回
    return WeatherData(temperature=22.5, humidity=60.0, condition=f"sunny in {city}")

# 二、定义结构化地理信息的TypedDict类型
class LocationInfo(TypedDict):
    # 纬度字段
    latitude: float
    # 经度字段
    longitude: float
    # 地点名称
    name: str

# 注册get_location为工具，根据地址返回结构化地理信息
@mcp.tool()
def get_location(address: str) -> LocationInfo:
    # 构造结构化地理坐标数据返回
    return LocationInfo(latitude=39.9042, longitude=116.4074, name=f"{address} 附近")

# 三、注册get_statistics为工具，返回结构化统计数据字典
@mcp.tool()
def get_statistics(data_type: str) -> dict[str, float]:
    # 如果数据类型为scores，返回分数统计
    if data_type == "scores":
        return {
            "mean": 88.6,
            "median": 90.0,
        }
    # 默认返回其他数据统计
    return {"mean": 42.5, "median": 40.0}

# 四、定义有类型注解的普通类，用户信息模型
class UserProfile:
    # 用户名字段
    name: str
    # 年龄字段
    age: int
    # 邮箱字段，可以为None
    email: str | None

    # 构造方法，初始化用户的各字段
    def __init__(self, name: str, age: int, email: str | None = None) -> None:
        self.name = name
        self.age = age
        self.email = email

# 注册get_user为工具，根据user_id返回结构化用户信息
@mcp.tool()
def get_user(user_id: str) -> UserProfile:
    # 如果为特定id，返回实例化用户
    if user_id == "u001":
        return UserProfile(name="Alice", age=30, email="alice@example.com")
    # 默认返回游客用户
    return UserProfile(name="Guest", age=0, email=None)

# 五、注册get_temperature为工具，返回单个浮点数温度（原始类型会自动包装为result字段）
@mcp.tool()
def get_temperature(city: str) -> float:
    # 返回温度数值
    return 21.7

# 注册list_cities为工具，返回城市名称列表
@mcp.tool()
def list_cities() -> list[str]:
    # 返回示例城市列表
    return ["Beijing", "Shanghai", "Shenzhen"]

# 六、定义不可结构化的类（属性无类型注解），用于模拟非结构化返回
class UntypedConfig:
    # 构造方法，保存参数到实例属性
    def __init__(self, setting1, setting2):  # type: ignore[reportMissingParameterType]
        self.setting1 = setting1
        self.setting2 = setting2

# 注册get_config为工具，返回非结构化配置对象
@mcp.tool()
def get_config() -> UntypedConfig:
    # 返回UntypedConfig示例对象
    return UntypedConfig("value1", "value2")

# 七、注册unstructured_message为工具，禁用结构化输出（返回字典也做非结构化内容处理）
@mcp.tool(structured_output=False)
def unstructured_message() -> dict[str, str]:
    # 返回一个普通字典，但不会被结构化解析
    return {"msg": "This will be returned as unstructured content."}

# 主程序入口，通过 stdio 方式启动 MCP 服务器
if __name__ == "__main__":
    if hasattr(sys.stdout, "reconfigure"):
        sys.stdout.reconfigure(encoding="utf-8")
        sys.stdin.reconfigure(encoding="utf-8")
    mcp.run(transport="stdio")

官方代码

# 工具结构化服务器示例，演示多种结构化输出方式与如何禁用结构化输出

# 导入 sys 模块，用于设置标准流编码
import sys
# 导入 TypedDict 用于类型提示
from typing import TypedDict

# 导入 Pydantic 的 BaseModel 和 Field，用于定义结构化数据模型
from pydantic import BaseModel, Field
d
# 从 mcp_lite 导入 FastMCP 服务器框架
from mcp.server.fastmcp import FastMCP

# 创建FastMCP服务器实例，设置服务名为"Tools Structured Output"
mcp = FastMCP(name="Tools Structured Output")

# 一、定义结构化天气数据模型，继承自Pydantic BaseModel
class WeatherData(BaseModel):
    # 温度字段，浮点型，带描述
    temperature: float = Field(description="摄氏温度")
    # 湿度字段，浮点型，带描述
    humidity: float = Field(description="湿度百分比")
    # 天气状况，字符串
    condition: str = Field(description="天气状况")

# 注册get_weather为工具，根据城市返回结构化天气数据
@mcp.tool()
def get_weather(city: str) -> WeatherData:
    # 构造结构化天气数据返回
    return WeatherData(temperature=22.5, humidity=60.0, condition=f"sunny in {city}")

# 二、定义结构化地理信息的TypedDict类型
class LocationInfo(TypedDict):
    # 纬度字段
    latitude: float
    # 经度字段
    longitude: float
    # 地点名称
    name: str

# 注册get_location为工具，根据地址返回结构化地理信息
@mcp.tool()
def get_location(address: str) -> LocationInfo:
    # 构造结构化地理坐标数据返回
    return LocationInfo(latitude=39.9042, longitude=116.4074, name=f"{address} 附近")

# 三、注册get_statistics为工具，返回结构化统计数据字典
@mcp.tool()
def get_statistics(data_type: str) -> dict[str, float]:
    # 如果数据类型为scores，返回分数统计
    if data_type == "scores":
        return {
            "mean": 88.6,
            "median": 90.0,
        }
    # 默认返回其他数据统计
    return {"mean": 42.5, "median": 40.0}

# 四、定义有类型注解的普通类，用户信息模型
class UserProfile:
    # 用户名字段
    name: str
    # 年龄字段
    age: int
    # 邮箱字段，可以为None
    email: str | None

    # 构造方法，初始化用户的各字段
    def __init__(self, name: str, age: int, email: str | None = None) -> None:
        self.name = name
        self.age = age
        self.email = email

# 注册get_user为工具，根据user_id返回结构化用户信息
@mcp.tool()
def get_user(user_id: str) -> UserProfile:
    # 如果为特定id，返回实例化用户
    if user_id == "u001":
        return UserProfile(name="Alice", age=30, email="alice@example.com")
    # 默认返回游客用户
    return UserProfile(name="Guest", age=0, email=None)

# 五、注册get_temperature为工具，返回单个浮点数温度（原始类型会自动包装为result字段）
@mcp.tool()
def get_temperature(city: str) -> float:
    # 返回温度数值
    return 21.7

# 注册list_cities为工具，返回城市名称列表
@mcp.tool()
def list_cities() -> list[str]:
    # 返回示例城市列表
    return ["Beijing", "Shanghai", "Shenzhen"]

# 六、定义不可结构化的类（属性无类型注解），用于模拟非结构化返回
class UntypedConfig:
    # 构造方法，保存参数到实例属性
    def __init__(self, setting1, setting2):  # type: ignore[reportMissingParameterType]
        self.setting1 = setting1
        self.setting2 = setting2

# 注册get_config为工具，返回非结构化配置对象
@mcp.tool()
def get_config() -> UntypedConfig:
    # 返回UntypedConfig示例对象
    return UntypedConfig("value1", "value2")

# 七、注册unstructured_message为工具，禁用结构化输出（返回字典也做非结构化内容处理）
@mcp.tool(structured_output=False)
def unstructured_message() -> dict[str, str]:
    # 返回一个普通字典，但不会被结构化解析
    return {"msg": "This will be returned as unstructured content."}

# 主程序入口，通过 stdio 方式启动 MCP 服务器
if __name__ == "__main__":
    if hasattr(sys.stdout, "reconfigure"):
        sys.stdout.reconfigure(encoding="utf-8")
        sys.stdin.reconfigure(encoding="utf-8")
    mcp.run(transport="stdio")

4. fastmcp.py #

mcp_lite/server/fastmcp.py

# 导入 sys 模块，用于标准输入输出操作
import sys
# 导入 base64 模块，用于二进制资源编码
import base64
# 导入 json 模块，用于 JSON 序列化
+import json
# 导入 re 模块，用于 URI 模板匹配
import re
# 从 urllib.parse 导入 unquote，用于 URL 解码
from urllib.parse import unquote
# 从 typing 导入 get_args、get_origin、get_type_hints
+from typing import get_args, get_origin, get_type_hints
# 从 mcp_lite.message 模块导入 SessionMessage 类
from mcp_lite.message import SessionMessage
# 从 mcp_lite.server 模块导入 stdio 子模块
from mcp_lite.server import stdio
# 导入 inspect 用于函数签名分析
import inspect
# 导入 asyncio 用于异步操作
import asyncio
# 从 mcp_lite.types 模块导入所有相关类型和类
from mcp_lite.types import (                          # 从 mcp_lite.types 模块导入以下类型和类
    JSONRPCRequest,                                   # JSONRPC 请求结构体
    JSONRPCError,                                     # JSONRPC 错误响应结构体
    ErrorData,                                        # 错误数据结构体
    InitializeResult,                                 # 初始化响应结构体
    LATEST_PROTOCOL_VERSION,                          # 最新协议版本常量
    ToolsCapability,                                  # 工具能力描述结构体
    ResourcesCapability,                              # 资源能力描述结构体
    ServerCapabilities,                               # 服务器能力结构体
    Implementation,                                   # 实现信息结构体
    JSONRPCResponse,                                  # JSONRPC 响应结构体
    ListToolsResult,                                  # 工具列表响应结构体
    Tool,                                             # 工具描述结构体
+   TextContent,                                      # 工具结果文本内容块
    CallToolRequestParams,                            # 调用工具请求参数结构体
    CallToolResult,                                   # 工具调用响应结构体
    Resource,                                         # 资源元数据结构体
    ResourceTemplate,                                 # 资源模板结构体
    ListResourcesResult,                              # 资源列表响应结构体
    ListResourceTemplatesResult,                      # 资源模板列表响应结构体
    ReadResourceRequestParams,                        # 读资源请求参数结构体
    ReadResourceResult,                               # 读资源响应结构体
    TextResourceContents,                             # 文本型资源内容结构体
    BlobResourceContents,                             # 二进制型资源内容结构体
)
# 从 pydantic 导入 BaseModel 并重命名为 PydanticBaseModel，便于后续判断数据模型类型
+from pydantic import BaseModel as PydanticBaseModel
# 从 typing_extensions 导入 is_typeddict 并重命名为 _is_typeddict，用于判断类型是否为 TypedDict
+from typing_extensions import is_typeddict as _is_typeddict

# 定义内部函数，根据函数的返回类型注解生成输出 schema，并确定是否需要包装输出
+def _output_schema_and_wrap(fn, structured_output: bool):
    # 如果未启用结构化输出，则直接返回 (None, False)
+   if not structured_output:
+       return None, False
+   try:
        # 获取函数签名
+       sig = inspect.signature(fn)
        # 获取返回注解类型
+       ann = sig.return_annotation
        # 如果没有返回注解，返回 (None, False)
+       if ann is inspect.Parameter.empty:
+           return None, False
+   except Exception:
        # 捕获异常，返回 (None, False)
+       return None, False
    # 调用 _schema_from_annotation 根据类型注解生成 schema 和 wrap_output
+   return _schema_from_annotation(ann, fn.__name__)

# 定义内部函数，根据类型注解生成 (output_schema, wrap_output)
+def _schema_from_annotation(ann, func_name: str):
    # 如果注解为空，直接返回 (None, False)
+   if ann is inspect.Parameter.empty:
+       return None, False
    # 获取类型的 origin（例如 list、dict 等）
+   origin = get_origin(ann)
    # 初始化 wrap 标志为 False
+   wrap = False
    # 初始化 schema
+   schema = None
    # 如果 ann 是 Pydantic BaseModel 的子类，则调用其 model_json_schema 方法生成 schema
+   if PydanticBaseModel and isinstance(ann, type) and issubclass(ann, PydanticBaseModel):
+       schema = ann.model_json_schema()
+       return schema, False
    # 如果 ann 有 __annotations__ 属性且不是 dict 或 list
+   if hasattr(ann, "__annotations__") and not (origin is dict or origin is list):
+       print(f"__annotations____annotations____annotations____annotations__: {ann}")
        # 判断是否为 TypedDict
+       if _is_typeddict(ann):
            # 获取 TypedDict 的类型提示
+           hints = get_type_hints(ann) if hasattr(ann, "__annotations__") else {}
            # 构建属性对应的 schema 字典
+           props = {}
+           for k, v in hints.items():
+               t = v
                # 判断类型分支，设置 JSON schema 类型
+               if t is int or t is type(None) and int in get_args(v):
+                   props[k] = {"type": "integer"}
+               elif t is float:
+                   props[k] = {"type": "number"}
+               elif t is str:
+                   props[k] = {"type": "string"}
+               elif t is bool:
+                   props[k] = {"type": "boolean"}
+               else:
+                   props[k] = {"type": "string"}
            # 构建最终对象类型 schema，全部属性都为必需
+           schema = {"type": "object", "properties": props, "required": list(props)}
+           return schema, False
        # 普通类（有类型注解）
+       try:
            # 获取类型提示
+           hints = get_type_hints(ann)
+           if hints:
                # 构建属性 schema
+               props = {}
+               for k, v in hints.items():
+                   if v is int or v is type(None):
+                       props[k] = {"type": "integer"}
+                   elif v is float:
+                       props[k] = {"type": "number"}
+                   elif v is str:
+                       props[k] = {"type": "string"}
+                   elif v is bool:
+                       props[k] = {"type": "boolean"}
+                   elif get_origin(v) is list:
+                       props[k] = {"type": "array", "items": {"type": "string"}}
+                   else:
+                       props[k] = {"type": "string"}
                # 返回对象类型 schema
+               schema = {"type": "object", "properties": props}
+               return schema, False
+       except Exception:
            # 获取类型提示失败时忽略
+           pass
    # 如果 ann 是 dict 类型
+   if origin is dict:
        # 获取泛型参数
+       args = get_args(ann)
        # 如果是 dict[str, X] 结构
+       if len(args) == 2 and args[0] is str:
+           vt = args[1]
+           if vt is float:
+               schema = {"type": "object", "additionalProperties": {"type": "number"}}
+           elif vt is int:
+               schema = {"type": "object", "additionalProperties": {"type": "integer"}}
+           elif vt is str:
+               schema = {"type": "object", "additionalProperties": {"type": "string"}}
+           else:
+               schema = {"type": "object", "additionalProperties": {}}
+           return schema, False
        # 否则需要 wrap
+       wrap = True
    # 如果 ann 是 list 或基本类型（int, float, str, bool, None），需要 wrap
+   if origin is list or ann in (int, float, str, bool, type(None)):
+       wrap = True
    # 如果需要 wrap，则将结果转换成 { "result": ... } 的包装结构
+   if wrap:
        # 结果默认为字符串类型
+       result_schema = {"type": "string"}
        # 针对不同类型调整 result 的类型
+       if ann is int:
+           result_schema = {"type": "integer"}
+       elif ann is float:
+           result_schema = {"type": "number"}
+       elif ann is bool:
+           result_schema = {"type": "boolean"}
+       elif origin is list:
+           result_schema = {"type": "array", "items": {"type": "string"}}
        # 返回包装对象 schema，并标记需要 wrap
+       schema = {"type": "object", "properties": {"result": result_schema}, "required": ["result"]}
+       return schema, True
    # 如果不满足上述情况，则返回 (None, False)
+   return None, False

# 定义内部函数，将工具函数返回内容转换为 structured_content 字典
+def _to_structured(out, output_schema, wrap_output):
    # 如果输出 schema 为空，直接返回 None
+   if output_schema is None:
+       return None
+   try:
        # 如果 out 是 PydanticBaseModel 的实例，则调用其 model_dump
+       if PydanticBaseModel and isinstance(out, PydanticBaseModel):
+           return out.model_dump(mode="json")
        # 如果需要 wrap_output，则将内容包装为 {"result": ...}
+       if wrap_output:
+           return {"result": out}
        # 如果 out 是字典类型，则直接返回 dict 拷贝
+       if isinstance(out, dict):
+           return dict(out)
        # 如果 out 有 __annotations__ 属性，则提取有注解的字段为字典
+       if hasattr(out, "__annotations__"):
+           hints = get_type_hints(type(out)) if hasattr(type(out), "__annotations__") else getattr(out, "__annotations__", {})
+           return {k: getattr(out, k) for k in hints if hasattr(out, k)}
        # 否则仍然包装为 {"result": ...}
+       return {"result": out}
+   except Exception:
        # 发生异常时，将内容以字符串形式包装返回
+       return {"result": str(out)}


# 内部函数：根据函数定义提取参数信息并生成 schema
def _schema(fn):
    # 获取函数签名
   sig = inspect.signature(fn)
    # 用于存储参数属性的字典
   props = {}
    # 用于存储必需参数名的列表
   req = []
    # 遍历所有参数
   for n, p in sig.parameters.items():
        # 跳过以 "_" 开头的参数
       if n.startswith("_"):
           continue
        # 所有参数类型都设为 string，title 为参数名
       props[n] = {"type": "string", "title": n}
        # 如果参数没有默认值，则为必需参数
       if p.default is inspect.Parameter.empty:
           req.append(n)
    # 返回对象类型 schema，包括属性和必需项目
   return {"type": "object", "properties": props, "required": req}
# 内部类：用于封装工具函数
class _Tool:
    # 初始化方法，接收目标函数、工具名、描述和结构化输出相关参数
+  def __init__(self, fn, name=None, desc=None, structured_output=True):
        # 保存原始函数引用
       self.fn = fn
        # 工具名称，优先采用传参，否则用函数名
       self.name = name or fn.__name__
        # 工具描述，优先采用传参，否则用函数 docstring
       self.desc = desc or (fn.__doc__ or "").strip()
        # 自动生成入参 schema
       self.schema = _schema(fn)
        # 判断函数是否为异步（协程）函数
       self.async_fn = asyncio.iscoroutinefunction(fn)
        # 是否启用结构化输出
+      self.structured_output = structured_output
        # 从返回类型注解生成 output_schema 和 wrap_output
+      self.output_schema, self.wrap_output = _output_schema_and_wrap(fn, structured_output) if structured_output else (None, False)

    # 转换为 Tool 类型的对象
   def to_tool(self):
        # 返回 Tool 类型实例，包含 output_schema
+      return Tool(name=self.name, description=self.desc or None, input_schema=self.schema, output_schema=self.output_schema)

    # 执行工具函数，参数为字典类型
   def run(self, args):
        # 如果是异步函数则使用 asyncio.run 执行，否则同步调用
       if self.async_fn:
           return asyncio.run(self.fn(**args))
       return self.fn(**args)


# 内部类：用于封装静态资源
class _Resource:
    # 构造方法，保存资源 URI、回调函数和 MIME 类型，检查函数是否异步
    def __init__(self, uri, fn, mime_type="text/plain"):
        # 保存资源 URI
        self.uri = uri
        # 保存资源内容生成函数
        self.fn = fn
        # 保存资源 MIME 类型
        self.mime_type = mime_type
        # 检查并保存函数是否为异步函数
        self.async_fn = asyncio.iscoroutinefunction(fn)

    # 执行资源内容生成函数，支持同步/异步
    def run(self):
        # 若为异步函数则通过 asyncio 运行
        if self.async_fn:
            return asyncio.run(self.fn())
        # 否则直接同步调用
        return self.fn()


# 内部类：用于封装模板资源
class _ResourceTemplate:
    # 构造方法，保存 URI 模板、处理函数、MIME 类型，并提取模板参数名
    def __init__(self, uri_template, fn, mime_type="text/plain"):
        # 保存 URI 模板
        self.uri_template = uri_template
        # 保存用于动态生成资源内容的函数
        self.fn = fn
        # 保存资源 MIME 类型
        self.mime_type = mime_type
        # 检查函数是否为异步函数
        self.async_fn = asyncio.iscoroutinefunction(fn)
        # 获取函数签名
        sig = inspect.signature(fn)
        # 提取所有非下划线开头的参数名，作为模板变量名
        self.param_names = [n for n in sig.parameters if not n.startswith("_")]

    # 匹配实际 URI 是否符合模板，并返回变量字典
    def matches(self, uri):
        # 使用正则表达式将 URI 模板转换成命名捕获组的正则模式
        pattern = self.uri_template.replace("{", "(?P<").replace("}", ">[^/]+)")
        # 尝试用给定的 URI 匹配该模式
        m = re.match(f"^{pattern}$", uri)
        # 若匹配成功则返回捕获的参数字典，进行 URL 解码
        if m:
            return {k: unquote(v) for k, v in m.groupdict().items()}
        # 匹配失败返回 None
        return None

    # 执行模板资源内容生成函数，支持异步/同步
    def run(self, args):
        # 若为异步函数，通过 asyncio 运行
        if self.async_fn:
            return asyncio.run(self.fn(**args))
        # 否则直接同步调用
        return self.fn(**args)


# 主服务类 FastMCP
class FastMCP:
    # 初始化方法，支持自定义服务器名称
    def __init__(self, name="mcp-server"):
        # 保存服务器名称
        self.name = name
        # 初始化工具池为字典
        self._tools = {}
        # 初始化静态资源池
        self._resources = {}
        # 初始化资源模板池
        self._resource_templates = {}

    # 工具注册装饰器，支持自定义工具名、描述和结构化输出
+   def tool(self, name=None, description=None, structured_output=True):
        # 装饰器实际函数
       def deco(fn):
            # 创建 _Tool 实例
+          t = _Tool(fn, name, description, structured_output=structured_output)
            # 注册进工具池
           self._tools[t.name] = t
            # 返回原始函数
           return fn
        # 返回装饰器
       return deco

    # 资源注册装饰器，支持静态资源和模板资源
    def resource(self, uri, mime_type="text/plain"):
        def deco(fn):
            if "{" in uri and "}" in uri:
                # 模板资源
                template = _ResourceTemplate(uri, fn, mime_type)
                self._resource_templates[uri] = template
            else:
                # 静态资源
                res = _Resource(uri, fn, mime_type)
                self._resources[uri] = res
            return fn
        return deco

    # 实际 JSONRPC 方法分发与业务处理
    def _handle(self, req):
        # 拆分 method、params、id
        method, params, rid = req.method, req.params or {}, req.id
        # 处理初始化请求
        if method == "initialize":
            # 组装初始化响应，包括协议版本、能力与服务信息
            caps = ServerCapabilities(
                tools=ToolsCapability(),
                resources=ResourcesCapability(),
            )
            r = InitializeResult(
                protocol_version=LATEST_PROTOCOL_VERSION,
                capabilities=caps,
                server_info=Implementation(name=self.name, version="0.1.0"),
            )
            # 返回 JSONRPCResponse 包装的结果
            return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))
         # 处理工具列表请求
        if method == "tools/list":
            # 构建工具列表结果
           r = ListToolsResult(tools=[t.to_tool() for t in self._tools.values()])
            # 返回 JSONRPCResponse 包装的结果
           return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))     
          # 处理工具调用请求
        # 如果方法是 "tools/call"
        if method == "tools/call":
            # 用 CallToolRequestParams 校验参数并实例化
+           p = CallToolRequestParams.model_validate(params, by_name=False)
            # 根据工具名称查找工具
+           t = self._tools.get(p.name)
            # 如果工具不存在，返回参数错误的 JSONRPCError
+           if not t:
+               return JSONRPCError(
+                   jsonrpc="2.0", 
+                   id=rid, 
+                   error=ErrorData(code=-32602, message=f"Unknown tool: {p.name}")
+               )
+           try:
                # 执行工具，传入参数（默认为空字典）
+               out = t.run(p.arguments or {})
                # 如果返回已经是 CallToolResult，直接赋值给 r
+               if isinstance(out, CallToolResult):
+                   r = out
+               else:
                    # 默认为无结构化内容
+                   struct = None
                    # 如果工具有定义 output_schema，则用 _to_structured 转为结构化内容
+                   if t.output_schema is not None:
+                       struct = _to_structured(out, t.output_schema, t.wrap_output)
                    # 如果输出是字符串，则用 TextContent 包装
+                   if isinstance(out, str):
+                       c = [TextContent(text=out)]
                    # 如果输出是 None，则内容列表为空
+                   elif out is None:
+                       c = []
                    # 如果有结构化内容，用 json.dumps 输出成文本
+                   elif struct is not None:
+                       c = [TextContent(text=json.dumps(struct, ensure_ascii=False, indent=2))]
+                   else:
+                       try:
                            # 如果输出是 PydanticBaseModel 实例，则用 model_dump 转字典，转 json 字符串
+                           if PydanticBaseModel and isinstance(out, PydanticBaseModel):
+                               text = json.dumps(out.model_dump(mode="json"), ensure_ascii=False, indent=2)
                            # 如果输出是字典，直接转 json
+                           elif isinstance(out, dict):
+                               text = json.dumps(out, ensure_ascii=False, indent=2)
                            # 其它类型转字符串
+                           else:
+                               text = str(out)
+                       except Exception:
                            # 如果序列化出错也转字符串
+                           text = str(out)
                        # 用文本内容包一层
+                       c = [TextContent(text=text)]
                    # 组装 CallToolResult，内容为 c，结构化内容为 struct
+                   r = CallToolResult(content=c, structured_content=struct)
+           except Exception as e:
                # 如果执行异常，返回错误内容，is_error 设为 True
+               r = CallToolResult(content=[TextContent(text=str(e))], is_error=True)
            # 返回 JSONRPCResponse，将结果 model_dump 并排除 None
+           return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))

        # 处理 resources/list 请求
        # 判断请求方法是否为 "resources/list"
        if method == "resources/list":
            # 遍历所有静态资源字典，将每个资源封装为 Resource 对象，组成资源列表
            resources = [
                Resource(uri=u, name=u, mime_type=r.mime_type)
                for u, r in self._resources.items()
            ]
            # 用资源列表生成 ListResourcesResult 响应体
            r = ListResourcesResult(resources=resources)
            # 将响应体封装为 JSONRPCResponse 并返回
            return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))

        # 处理 resources/templates/list 请求
        # 判断请求方法是否为 "resources/templates/list"
        if method == "resources/templates/list":
            # 遍历所有资源模板字典，将每个模板封装为 ResourceTemplate 对象，组成模板列表
            templates = [
                ResourceTemplate(uri_template=u, name=u, mime_type=t.mime_type)
                for u, t in self._resource_templates.items()
            ]
            # 用模板列表生成 ListResourceTemplatesResult 响应体
            r = ListResourceTemplatesResult(resource_templates=templates)
            # 封装为 JSONRPCResponse 并返回
            return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))

        # 处理 resources/read 请求
        # 判断请求方法是否为 "resources/read"
        if method == "resources/read":
            # 用 ReadResourceRequestParams 校验并实例化参数
            p = ReadResourceRequestParams.model_validate(params, by_name=False)
            # 将 uri 转为字符串
            uri_str = str(p.uri)
            try:
                # 先查找静态资源
                if res := self._resources.get(uri_str):
                    # 调用静态资源的 run 方法获取输出内容
                    out = res.run()
                    # 封装为 TextResourceContents（文本类型资源内容）
                    content = TextResourceContents(uri=uri_str, text=str(out), mime_type=res.mime_type)
                    # 构造 ReadResourceResult 响应体
                    r = ReadResourceResult(contents=[content])
                    # 封装为 JSONRPCResponse 并返回
                    return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))
                # 如果未找到静态资源，则依次查找所有资源模板
                for template in self._resource_templates.values():
                    # template.matches 返回匹配参数字典，匹配成功则进入分支
                    if args := template.matches(uri_str):
                        # 执行模板资源的 run 方法，传入提取的参数
                        out = template.run(args)
                        # 如果输出为 bytes 类型，封装为二进制内容
                        if isinstance(out, bytes):
                            content = BlobResourceContents(
                                uri=uri_str,
                                blob=base64.b64encode(out).decode(),
                                mime_type=template.mime_type or "application/octet-stream",
                            )
                        else:
                            # 其它类型一律作为文本类型内容封装
                            content = TextResourceContents(
                                uri=uri_str,
                                text=str(out),
                                mime_type=template.mime_type or "text/plain",
                            )
                        # 构造 ReadResourceResult 响应体
                        r = ReadResourceResult(contents=[content])
                        # 封装为 JSONRPCResponse 并返回
                        return JSONRPCResponse(jsonrpc="2.0", id=rid, result=r.model_dump(by_alias=True, exclude_none=True))
                # 静态资源和模板资源都未命中，返回资源不存在的错误
                return JSONRPCError(jsonrpc="2.0", id=rid, error=ErrorData(code=-32602, message=f"Unknown resource: {uri_str}"))
            except Exception as e:
                # 发生异常，返回内部错误（服务端错误，code -32603）
                return JSONRPCError(jsonrpc="2.0", id=rid, error=ErrorData(code=-32603, message=str(e)))

        # 不支持的方法，返回 "方法未找到" 错误（code -32601）
        return JSONRPCError(jsonrpc="2.0", id=rid, error=ErrorData(code=-32601, message=f"Method not found: {method}"))    
    # 处理 SessionMessage 消息，主要路由 JSONRPCRequest
    def _handle_msg(self, msg):
        # 判断消息类型不是 SessionMessage 则忽略
        if not isinstance(msg, SessionMessage):
            return None
        # 获取消息体
        m = msg.message
        # 只处理 JSONRPCRequest 类型（通知类消息如 notifications/initialized 不响应）
        if not isinstance(m, JSONRPCRequest) or getattr(m, "id", None) is None:
            return None
        print("[Server] Request:", m.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)    
        try:
            # 调用实际业务处理方法
            resp = self._handle(m)
            print("[Server] Response:", resp.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)
            return resp
        except Exception as e:
            # 捕获异常，返回标准 JSONRPCError（code -32603，内部错误）
            err = JSONRPCError(jsonrpc="2.0", id=m.id, error=ErrorData(code=-32603, message=str(e)))
            print("[Server] Response (error):", err.model_dump_json(by_alias=True, exclude_unset=True), file=sys.stderr)
            return err
    # 运行服务方法，默认采用 stdio 传输方式
    def run(self, transport="stdio"):
        # 仅支持 stdio，其他方式抛出异常
        if transport != "stdio":
            raise ValueError(f"unsupported transport: {transport}")
        # 启动 stdio server，当前对象 _handle_msg 作为消息处理回调
        stdio.stdio_server(self._handle_msg)    

5. types.py #

mcp_lite/types.py

# 从 pydantic 导入 BaseModel、ConfigDict、Field、TypeAdapter、field_validator
+from pydantic import BaseModel, ConfigDict, Field, TypeAdapter, field_validator
# 导入 pydantic 的 to_camel 驼峰命名生成器
from pydantic.alias_generators import to_camel
# 导入 Any 和 Literal 类型注解
from typing import Any, Literal
# 定义 RequestId 类型，既可以是 int 也可以是 str
RequestId = int | str
# 定义 MCP 的基础模型类，支持驼峰命名和按名称填充
class MCPModel(BaseModel):
    # 指定 Pydantic 模型配置：使用驼峰形式命名和按名称填充
   model_config = ConfigDict(alias_generator=to_camel, populate_by_name=True)
# 定义客户端能力结构体
class ClientCapabilities(MCPModel):
    # 可选字段：客户端实验性能力扩展，键为 str，值为字典
   experimental: dict[str, dict[str, Any]] | None = None
# 定义服务器或客户端的实现信息结构体
class Implementation(MCPModel):
    # 实现的名称
   name: str = ""
    # 实现的版本号
   version: str = ""
    # 可选字段：人类可读的标题
   title: str | None = None
    # 可选字段：实现的描述
   description: str | None = None    
# 定义初始化请求参数结构体
class InitializeRequestParams(MCPModel):
    # 协议版本号
   protocol_version: str = ""
    # 可选字段：客户端能力描述
   capabilities: ClientCapabilities = None
    # 可选字段：客户端实现信息
   client_info: Implementation = None
# 定义初始化请求结构体
class InitializeRequest(MCPModel):
    # 方法名，固定为 "initialize"
   method: Literal["initialize"] = "initialize"
    # 可选字段：参数，类型为 InitializeRequestParams
   params: InitializeRequestParams = None    
# 当前协议的最新版本号
LATEST_PROTOCOL_VERSION = "2024-11-05"
# 定义工具相关能力结构体
class ToolsCapability(MCPModel):
    # 工具列表是否发生变化，可选布尔类型
   list_changed: bool | None = None
# 定义资源相关能力结构体
class ResourcesCapability(MCPModel):
    # 是否支持资源订阅
   subscribe: bool | None = None
    # 资源列表是否变化通知
   list_changed: bool | None = None
# 定义服务端能力描述结构体
class ServerCapabilities(MCPModel):
    # 可选字段：实验性能力扩展
   experimental: dict[str, dict[str, Any]] | None = None
    # 可选字段：工具能力
   tools: ToolsCapability | None = None
    # 可选字段：资源能力
   resources: ResourcesCapability | None = None
# 定义初始化响应结构体
class InitializeResult(MCPModel):
    # 协议版本号
   protocol_version: str = ""
    # 可选字段：服务端能力描述
   capabilities: ServerCapabilities = None
    # 可选字段：服务端实现信息
   server_info: Implementation = None
    # 可选字段：初始化说明信息
   instructions: str | None = None
# 定义客户端初始化完成通知结构体
class InitializedNotification(MCPModel):
    # 方法名，固定为 "notifications/initialized"
   method: Literal["notifications/initialized"] = "notifications/initialized"
    # 可选字段：通知参数，可以为字典或 None
   params: dict[str, Any] | None = None    
# 定义 JSONRPC 请求的数据结构
class JSONRPCRequest(BaseModel):
    # jsonrpc 协议版本，固定为 "2.0"
    jsonrpc: Literal["2.0"] = "2.0"
    # 请求 ID，可以为 int 或 str 类型
    id: RequestId = None
    # 方法名称，字符串类型
    method: str = ""
    # 方法参数，为一个字典或 None
    params: dict[str, Any] | None = None

# 定义 JSONRPC 通知的数据结构（没有 id 字段）
class JSONRPCNotification(BaseModel):
    # jsonrpc 协议版本，固定为 "2.0"
    jsonrpc: Literal["2.0"] = "2.0"
    # 通知的方法名称
    method: str = ""
    # 通知参数，可以为字典或者 None
    params: dict[str, Any] | None = None

# 定义 JSONRPC 响应的数据结构
class JSONRPCResponse(BaseModel):
    # jsonrpc 协议版本，固定为 "2.0"
    jsonrpc: Literal["2.0"] = "2.0"
    # 响应的 ID，需要与请求 ID 匹配
    id: RequestId = None
    # 响应结果，可以为字典或 None
    result: dict[str, Any] = None

# 定义错误的数据结构
class ErrorData(BaseModel):
    # 错误码，默认值为 0
    code: int = 0
    # 错误信息，默认值为空字符串
    message: str = ""
    # 附加的错误数据，可以为任意类型或 None
    data: Any = None

# 定义 JSONRPC 错误消息的数据结构
class JSONRPCError(BaseModel):
    # jsonrpc 协议版本，固定为 "2.0"
    jsonrpc: Literal["2.0"] = "2.0"
    # 错误对应的请求 ID，可以为 None
    id: RequestId | None = None
    # 错误的详细信息，类型为 ErrorData
    error: ErrorData = None
# 定义所有 JSONRPC 消息的联合类型
JSONRPCMessage = JSONRPCRequest | JSONRPCNotification | JSONRPCResponse | JSONRPCError
# 定义 JSONRPC 消息适配器，用于类型自动推断和校验
jsonrpc_message_adapter = TypeAdapter(JSONRPCMessage)
# 定义工具结果中的文本内容块
+class TextContent(MCPModel):
+   type: Literal["text"] = "text"
+   text: str = ""

# 定义工具描述数据结构体
class Tool(MCPModel):
    # 工具名称
   name: str = ""
    # 可选字段：工具描述
   description: str | None = None
    # 工具输入参数的 schema，默认为空字典
   input_schema: dict[str, Any] = Field(default_factory=dict)
    # 可选字段：工具输出 schema
   output_schema: dict[str, Any] | None = None
# 定义获取工具列表请求结构体
class ListToolsRequest(MCPModel):
    # 方法名，固定为 "tools/list"
   method: Literal["tools/list"] = "tools/list"
    # 可选字段：参数，可以为字典或 None
   params: dict[str, Any] | None = None

# 定义获取工具列表响应结构体
class ListToolsResult(MCPModel):
    # 工具列表，默认为空列表
   tools: list[Tool] = []
    # 可选字段：分页游标，可为 None
   next_cursor: str | None = None

# 定义资源元数据结构体
class Resource(MCPModel):
    # 资源 URI
   uri: str = ""
    # 可选：资源名称
   name: str | None = None
    # 可选：人类可读标题
   title: str | None = None
    # 可选：资源描述
   description: str | None = None
    # 可选：MIME 类型
   mime_type: str | None = None

# 定义资源模板结构体
class ResourceTemplate(MCPModel):
    # URI 模板，如 greeting://{name}
   uri_template: str = ""
    # 可选：模板名称
   name: str | None = None
    # 可选：人类可读标题
   title: str | None = None
    # 可选：模板描述
   description: str | None = None
    # 可选：MIME 类型
   mime_type: str | None = None

# 定义 resources/list 请求结构体
class ListResourcesRequest(MCPModel):
   method: Literal["resources/list"] = "resources/list"
   params: dict[str, Any] | None = None

# 定义 resources/list 响应结构体
class ListResourcesResult(MCPModel):
   resources: list[Resource] = []
   next_cursor: str | None = None

# 定义 resources/templates/list 请求结构体
class ListResourceTemplatesRequest(MCPModel):
   method: Literal["resources/templates/list"] = "resources/templates/list"
   params: dict[str, Any] | None = None

# 定义 resources/templates/list 响应结构体
class ListResourceTemplatesResult(MCPModel):
   resource_templates: list[ResourceTemplate] = []
   next_cursor: str | None = None

# 定义 resources/read 请求参数结构体
class ReadResourceRequestParams(MCPModel):
   uri: str = ""

# 定义 resources/read 请求结构体
class ReadResourceRequest(MCPModel):
   method: Literal["resources/read"] = "resources/read"
   params: ReadResourceRequestParams = None

# 定义资源内容基类（文本）
class TextResourceContents(MCPModel):
   uri: str = ""
   mime_type: str | None = None
   text: str = ""

# 定义资源内容基类（二进制）
class BlobResourceContents(MCPModel):
   uri: str = ""
   mime_type: str | None = None
   blob: str = ""  # base64 编码

# 定义 resources/read 响应结构体
class ReadResourceResult(MCPModel):
   contents: list[TextResourceContents | BlobResourceContents] = []

# 定义调用工具请求参数结构体
class CallToolRequestParams(MCPModel):
    # 工具名称
   name: str = ""
    # 可选字段：输入参数，为字典类型或 None
   arguments: dict[str, Any] | None = None        
# 定义调用工具请求结构体
class CallToolRequest(MCPModel):
    # 方法名，固定为 "tools/call"
   method: Literal["tools/call"] = "tools/call"
    # 可选字段：参数，类型为 CallToolRequestParams
   params: CallToolRequestParams = None
# 定义调用工具响应结构体
class CallToolResult(MCPModel):
    # 内容字段，由 TextContent 或字典组成的列表，默认为空列表
+   content: list[TextContent | dict[str, Any]] = Field(default_factory=list)
    # 结构化内容字段，可为字典或 None
+   structured_content: dict[str, Any] | None = None
    # 错误标志字段，标识是否为错误结果，默认为 False
+   is_error: bool = False

    # 针对 content 字段的字段校验器，模型初始化前调用
+   @field_validator("content", mode="before")
+   @classmethod
+   def _parse_content(cls, v):
        # 如果传入的值不是列表，直接返回
+       if not isinstance(v, list):
+           return v
        # 初始化输出列表
+       out = []
        # 遍历每一项
+       for item in v:
            # 如果项是字典且类型为 "text"，转换为 TextContent 实例
+           if isinstance(item, dict) and item.get("type") == "text":
+               out.append(TextContent(text=item.get("text", "")))
            # 否则，原样加入输出列表
+           else:
+               out.append(item)
        # 返回处理后的列表
+       return out       

6. 工作流程 #

6.1 整体目标 #

在 MCP 工具调用中支持 结构化输出（Structured Output），使工具返回的数据既有人类可读的文本（content），又有可被程序解析的结构化数据（structured_content），并附带 output_schema 供客户端校验。

6.2 各模块修改说明 #

6.2.1. mcp_lite/types.py #

新增 TextContent：

class TextContent(MCPModel):
    type: Literal["text"] = "text"
    text: str = ""

用于表示工具结果中的文本块，对应 MCP 协议中的 type: "text" 内容。

修改 Tool：

• 增加 output_schema 字段，用于描述工具返回值的 JSON Schema。

修改 CallToolResult：

• content：list[TextContent | dict]，支持解析并转为 TextContent
• structured_content：可选结构化内容字典
• _parse_content：在解析前将 {"type":"text","text":"..."} 转为 TextContent，便于客户端用 isinstance(block, types.TextContent) 判断

说明： 文档中有一行调试打印 print(f"__annotations__...")，建议在正式代码中删除。

6.2.2. mcp_lite/server/fastmcp.py #

_output_schema_and_wrap(fn, structured_output)：

• 从函数返回类型注解生成 output_schema 和 wrap_output
• structured_output=False 时直接返回 (None, False)
• 无返回注解时返回 (None, False)

_schema_from_annotation(ann, func_name)：

按返回类型生成 schema 和 wrap 标志：

_to_structured(out, output_schema, wrap_output)：

将工具返回值转为 structured_content 字典：

• Pydantic 模型 → model_dump(mode="json")
• wrap_output=True → {"result": out}
• 字典 → 直接返回
• 带 __annotations__ 的普通类 → 按注解字段提取为字典

_Tool 扩展：

• 新增 structured_output、output_schema、wrap_output
• to_tool() 在 Tool 中带上 output_schema

tool() 装饰器：

• 增加 structured_output=True 参数
• `@mcp.tool(structured_output=False)` 可关闭结构化输出

tools/call 处理：

• 执行工具后：out = t.run(...)
• 若 t.output_schema 存在，则 struct = _to_structured(out, ...)
• 构建 content：str → TextContent(text=out)；struct 存在 → TextContent(text=json.dumps(struct))；否则按 str(out) 或 JSON 序列化
• 返回 CallToolResult(content=c, structured_content=struct)

6.2.3. tools_structured_client.py #

• 使用 mcp_lite 的同步接口
• print_call_result 同时处理 content 和 structured_content：
  • 支持 isinstance(block, types.TextContent) 和 block.get("type")=="text"

6.2.4. tools_structured_server.py #

• 从 mcp.server.fastmcp 改为 mcp_lite.server.fastmcp
• 启动前设置 sys.stdout.reconfigure(encoding="utf-8") 避免中文乱码

6.3 数据流概览 #

工具注册 @mcp.tool(structured_output=True)
    → _Tool 解析返回类型注解
    → _schema_from_annotation 生成 output_schema、wrap_output
    → tools/list 返回 Tool(output_schema=...)

客户端调用 call_tool("get_weather", {"city":"Hangzhou"})
    → 服务端 t.run() 得到 WeatherData 实例
    → _to_structured(out) → {"temperature":22.5, "humidity":60, "condition":"sunny in Hangzhou"}
    → content = [TextContent(text=json.dumps(struct))]
    → structured_content = struct
    → 客户端解析 contents 和 structured_content

6.4 结构化输出类型映射 #

6.5 要点总结 #

1. 双输出：content 为可读文本，structured_content 为可解析结构。
2. schema 驱动：通过返回类型注解自动生成 output_schema。
3. 类型覆盖：支持 BaseModel、TypedDict、普通类、dict、list、基本类型。
4. 可关闭：structured_output=False 时只返回文本，不生成结构化内容。