1. 结构化输出（Structured Output） #

本章主要介绍如何使用 MCP 的结构化输出机制，包括定义不同形式的结构化输出值，以及客户端如何获取和解析结构化输出值。你将学会：

• 如何使用 Pydantic 定义结构化输出值；
• 客户端如何获取和解析结构化输出值；
• 如何使用结构化输出进行多轮对话。

2. 服务器 #

新建 C:\mcp-project\tools_structured_server.py：

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

# 导入类型提示相关模块：TypedDict、Dict、List
from typing import TypedDict

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

# 导入FastMCP高层服务器
from mcp.server.fastmcp import FastMCP

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


# 一、定义Pydantic BaseModel用于返回结构化天气数据
class WeatherData(BaseModel):
    # 天气数据的结构定义（Pydantic模型）
    temperature: float = Field(description="摄氏温度")
    humidity: float = Field(description="湿度百分比")
    condition: str = Field(description="天气状况")


# 注册为工具：根据输入城市返回结构化天气数据
@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


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


# 三、返回dict[str, float]等可推导Schema的类型
@mcp.tool()
def get_statistics(data_type: str) -> dict[str, float]:
    # 返回统计数据（结构化字典）
    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


# 注册为工具：根据user_id返回用户信息
@mcp.tool()
def get_user(user_id: str) -> UserProfile:
    # 返回用户信息（结构化）
    # 根据user_id造一个用户
    if user_id == "u001":
        return UserProfile(name="Alice", age=30, email="alice@example.com")
    return UserProfile(name="Guest", age=0, email=None)


# 五、返回原始类型与列表类型：将被自动包装为{"result": ...}
# 注册为工具：返回温度（原始浮点数）
@mcp.tool()
def get_temperature(city: str) -> float:
    # 返回温度（原始浮点数，客户端会在structuredContent.result中看到）
    return 21.7


# 注册为工具：返回城市列表
@mcp.tool()
def list_cities() -> list[str]:
    # 返回城市列表（列表会被包装到structuredContent.result中）
    return ["Beijing", "Shanghai", "Shenzhen"]


# 六、定义不可序列化的类（无类型注解字段）：将被视为非结构化
class UntypedConfig:
    # 无类型注解的属性：无法生成结构化Schema，会退化为非结构化文本内容
    def __init__(self, setting1, setting2):  # type: ignore[reportMissingParameterType]
        self.setting1 = setting1
        self.setting2 = setting2


# 注册为工具：返回非结构化配置对象
@mcp.tool()
def get_config() -> UntypedConfig:
    # 返回非结构化配置对象（将作为文本内容返回）
    return UntypedConfig("value1", "value2")


# 七、禁用结构化输出：即便返回dict也按非结构化处理
@mcp.tool(structured_output=False)
def unstructured_message() -> dict[str, str]:
    # 关闭结构化输出：返回字典也当作非结构化文本
    return {"msg": "This will be returned as unstructured content."}


# 主入口：以stdio方式运行服务器
if __name__ == "__main__":
    mcp.run(transport="stdio")

说明：

• 返回类型带有清晰的类型注解时，FastMCP 会生成 outputSchema 并验证结果。
• 原始类型与列表/元组等会被自动包装在 {"result": ...} 中。
• 无法推断的类（无类型注解）将退化为“非结构化”文本内容返回。
• 通过 `@mcp.tool(structured_output=False)` 可显式关闭结构化返回。

3. 客户端 #

新建 C:\mcp-project\test_client_tools_structured.py：

# 导入异步IO库，用于异步编程
import asyncio

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

# 从mcp模块导入客户端会话、Stdio参数和类型定义
from mcp import ClientSession, StdioServerParameters, types

# 从mcp.client.stdio导入stdio客户端工厂方法
from mcp.client.stdio import stdio_client


# 定义辅助函数，用于打印工具调用结果
def print_call_result(tag: str, result: types.CallToolResult) -> None:
    """辅助打印工具调用结果：同时查看 content 与 structuredContent。"""
    # 初始化可读内容列表
    readable_parts: list[str] = []
    # 遍历结果中的content部分
    for block in result.content:
        # 如果内容块是文本类型，则提取文本
        if isinstance(block, types.TextContent):
            readable_parts.append(block.text)
    # 将所有文本内容用" | "拼接，若无内容则显示<no text content>
    printable_text = (
        " | ".join(readable_parts) if readable_parts else "<no text content>"
    )

    # 获取结构化内容（structuredContent），若无则为None
    structured = getattr(result, "structuredContent", None)

    # 打印文本内容
    print(f"[{tag}] text=", printable_text)
    # 打印结构化内容
    print(f"[{tag}] structured=", structured)


# 定义主异步函数
async def main() -> None:
    # 1) 获取当前文件的绝对路径所在目录
    base_dir = os.path.dirname(os.path.abspath(__file__))
    # 2) 拼接出服务器脚本的绝对路径
    server_path = os.path.join(base_dir, "tools_structured_server.py")

    # 3) 配置以stdio方式启动服务器的参数
    server_params = StdioServerParameters(
        command="python",
        args=[server_path],
        env={},
    )

    # 4) 使用stdio_client建立与服务器的连接
    async with stdio_client(server_params) as (read, write):
        # 5) 创建客户端会话
        async with ClientSession(read, write) as session:
            # 6) 初始化会话，完成握手
            await session.initialize()

            # 7) 列出所有可用工具，确认注册情况
            tools = await session.list_tools()
            print("[Tools]", [t.name for t in tools.tools])

            # 8) 依次调用不同的工具，并打印结果

            # 调用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工具，查询用户u001信息
            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工具，获取配置信息（非结构化，主要在text content中）
            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())

说明：

• CallToolResult.content：面向可读性的内容块列表（例如文本）。
• CallToolResult.structuredContent：机器可读的结构化结果；当返回类型可生成 outputSchema 时会出现。
• 对原始值与序列（如 float, list[str]），会看到 structuredContent = {"result": ...}。

4. 运行与验证 #

cd C:\mcp-project
call .venv\Scripts\activate
python test_client_tools_structured.py

预期输出（示例，字段顺序可能不同）：

[Tools] ['get_weather', 'get_location', 'get_statistics', 'get_user', 'get_temperature', 'list_cities', 'get_config', 'unstructured_message']
[get_weather] text= {
  "temperature": 22.5,
  "humidity": 60.0,
  "condition": "sunny in Hangzhou"
}
[get_weather] structured= {'temperature': 22.5, 'humidity': 60.0, 'condition': 'sunny in Hangzhou'}
[get_location] text= {
  "latitude": 39.9042,
  "longitude": 116.4074,
  "name": "天安门 附近"
}
[get_location] structured= {'latitude': 39.9042, 'longitude': 116.4074, 'name': '天安门 附近'}
[get_statistics] text= {
  "mean": 88.6,
  "median": 90.0
}
[get_statistics] structured= {'mean': 88.6, 'median': 90.0}
[get_user] text= "<__main__.UserProfile object at 0x00000203E4DD3B60>"
[get_user] structured= {'name': 'Alice', 'age': 30, 'email': 'alice@example.com'}
[get_temperature] text= 21.7
[get_temperature] structured= {'result': 21.7}
[list_cities] text= Beijing | Shanghai | Shenzhen
[list_cities] structured= {'result': ['Beijing', 'Shanghai', 'Shenzhen']}
[get_config] text= "<__main__.UntypedConfig object at 0x00000203E4DD3B60>"
[get_config] structured= None
[unstructured_message] text= {
  "msg": "This will be returned as unstructured content."
}
[unstructured_message] structured= None

若输出与示例相近，表示你已掌握结构化输出的核心用法与客户端解析方式。

5.概念 #

特性	TypedDict	BaseModel	Field
类型检查	静态类型提示	运行时验证	字段配置
验证	无	内置验证器	验证规则
序列化	手动处理	自动支持	配置选项
运行时	普通字典	模型实例	元数据
性能	零开销	验证开销	配置开销

5.1. TypedDict - 类型化字典 #

TypedDict 是 Python 3.8+ 引入的类型提示功能，用于定义字典的结构和类型。

from typing import TypedDict, Optional, List

# 基础 TypedDict
class UserInfo(TypedDict):
    name: str
    age: int
    email: str
    is_active: bool

# 使用示例
user: UserInfo = {
    "name": "张三",
    "age": 25,
    "email": "zhangsan@example.com",
    "is_active": True
}

# 嵌套 TypedDict
class Address(TypedDict):
    street: str
    city: str
    postal_code: str

class DetailedUser(TypedDict):
    user: UserInfo
    address: Address
    tags: List[str]

# 可选字段
class UserProfile(TypedDict, total=False):
    name: str
    age: int
    bio: Optional[str]  # 可选字段
    avatar: Optional[str]

TypedDict 的优势：

• 类型安全：IDE 提供自动补全和类型检查
• 文档化：代码即文档，清晰表达数据结构
• 运行时兼容：仍然是普通字典，向后兼容

5.2 BaseModel - Pydantic 数据模型 #

BaseModel 是 Pydantic 库的核心，提供数据验证、序列化和反序列化功能。

# 导入Pydantic的BaseModel、Field和validator，用于定义数据模型和字段验证
from pydantic import BaseModel, Field, ValidationError, field_validator

# 导入datetime用于时间字段
from datetime import datetime

# 导入List和Optional类型提示
from typing import List, Optional


# 定义用户数据模型，继承自BaseModel
class User(BaseModel):
    # 用户ID，整数类型
    id: int
    # 用户名，字符串类型，长度1-50，必填
    name: str = Field(..., min_length=1, max_length=50)
    # 邮箱，字符串类型，必须符合正则表达式格式
    email: str = Field(..., pattern=r"^[^@]+@[^@]+\.[^@]+$")
    # 年龄，可选整数，范围0-150
    age: Optional[int] = Field(None, ge=0, le=150)
    # 创建时间，默认为当前时间
    created_at: datetime = Field(default_factory=datetime.now)

    # 定义name字段的自定义验证器
    @field_validator("name")
    def name_must_be_valid(cls, v):
        # 去除空白后判断是否为空
        if not v.strip():
            raise ValueError("姓名不能为空")
        # 返回去除空白后的字符串
        return v.strip()

    # 配置类
    class Config:
        # 赋值时自动验证字段
        validate_assignment = True
        # 禁止出现未声明的额外字段
        extra = "forbid"


# 使用示例
try:
    # 创建User对象，传入各字段
    user = User(id=1, name="李四", email="lisi@example.com", age=30)
    # 打印为字典格式
    print(user.model_dump())
    # 打印为JSON格式
    print(user.model_dump_json())
# 捕获验证异常
except ValidationError as e:
    # 打印验证失败信息
    print(f"验证失败: {e}")

BaseModel 特性：

• 自动验证：类型、格式、范围等验证
• 类型转换：自动转换兼容类型
• 序列化：支持 JSON、字典等格式
• 嵌套模型：支持复杂数据结构

5.3. Field - 字段配置和验证 #

Field 用于配置字段的验证规则、默认值、描述等元数据。

# 导入Pydantic的BaseModel和Field，用于定义数据模型和字段属性
from pydantic import BaseModel, Field

# 导入List和Literal类型，用于类型注解
from typing import List, Literal


# 定义产品数据模型，继承自BaseModel
class Product(BaseModel):
    # 产品唯一标识符，必须提供
    id: int = Field(..., description="产品唯一标识符")
    # 产品名称，长度1-100，必须提供
    name: str = Field(..., min_length=1, max_length=100, description="产品名称")
    # 产品价格，必须大于0，必须提供
    price: float = Field(..., gt=0, description="产品价格（必须大于0）")
    # 产品描述，最大长度1000，默认为空字符串
    description: str = Field("", max_length=1000, description="产品描述")
    # 产品标签，字符串列表，最多10个标签，默认为空列表
    tags: List[str] = Field(default_factory=list, max_items=10, description="产品标签")
    # 库存数量，默认为0，不能为负数
    stock: int = Field(0, ge=0, description="库存数量")

    # 产品编码，使用alias“code”进行字段映射，必须提供
    product_code: str = Field(..., alias="code", description="产品编码")

    # 产品状态，只能为"active"，使用Literal类型限定
    status: Literal["active"] = Field("active", description="产品状态")

    # SKU编码，必须符合正则表达式格式：2个大写字母+6位数字
    sku: str = Field(..., pattern=r"^[A-Z]{2}\d{6}$", description="SKU编码")


# 创建Product实例，传入各字段参数
product = Product(
    id=1,
    name="智能手机",
    price=2999.99,
    description="高性能智能手机",
    tags=["电子", "通讯"],
    code="PH001",  # 通过alias“code”传递产品编码
    sku="PH123456",  # 必须的SKU字段，符合格式
)

# 使用别名（如code）输出模型数据
print(product.model_dump(by_alias=True))
# 输出分隔符
print("\n=== 字段访问 ===")
# 通过原始字段名访问产品编码
print(f"product_code: {product.product_code}")
# 访问SKU字段
print(f"sku: {product.sku}")

Field 常用参数：

• default: 默认值
• alias: 字段别名
• description: 字段描述
• min_length/max_length: 字符串长度限制
• gt/ge/lt/le: 数值范围限制
• regex: 正则表达式验证
• const: 常量值

5.4. structuredContent #

structuredContent 是 MCP 协议中用于传递结构化数据的字段，通常与 content 字段配合使用。

structuredContent 的作用是让 AI 能够以结构化的方式理解和处理内容，而不是仅仅处理纯文本。它的主要作用包括：

1. 结构化理解

• 语义层次：AI 可以理解内容的逻辑结构（标题、段落、列表、代码块等）
• 上下文关系：能够识别不同部分之间的关联性和层次关系
• 内容类型：区分普通文本、代码、表格、图片等不同类型的内容

2. 更精确的处理

• 定位准确：可以精确定位到特定的内容片段或结构元素
• 上下文感知：在回答问题时能够考虑完整的上下文结构
• 引用精确：能够准确引用文档中的特定部分

3. 更好的用户体验

• 回答质量：AI 的回答更加准确和有条理
• 引用清晰：能够明确指出信息来源和具体位置
• 逻辑连贯：回答的逻辑结构更清晰

4. 技术优势

• 解析效率：比纯文本解析更高效
• 错误减少：减少对内容结构的误解
• 扩展性：支持更复杂的内容类型和交互

# 导入异步IO库
import asyncio

# 导入日期时间模块
from datetime import datetime

# 导入类型注解相关
from typing import List, Optional, Any, Literal

# 导入pydantic的数据模型基类和字段定义
from pydantic import BaseModel, Field

# MCP 相关导入
from mcp.server.fastmcp import FastMCP
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.stdio import stdio_client


# 1. MCP 响应结构定义
# 定义内容块基类，继承自pydantic的BaseModel
class ContentBlock(BaseModel):
    """内容块基类"""

    # 占位，不添加实际字段
    pass


# 定义文本内容块，继承自ContentBlock
class TextContent(ContentBlock):
    """文本内容块"""

    # 类型字段，固定为"text"
    type: Literal["text"] = "text"
    # 文本内容
    text: str


# 定义数据内容块，继承自ContentBlock
class DataContent(ContentBlock):
    """数据内容块"""

    # 类型字段，固定为"data"
    type: Literal["data"] = "data"
    # 任意类型的数据
    data: Any
    # 数据的MIME类型
    mimeType: str


# 定义MCP响应结构
class MCPResponse(BaseModel):
    """MCP 响应结构"""

    # 可读内容块列表
    content: List[ContentBlock]  # 可读内容
    # 可选的结构化内容
    structuredContent: Optional[Any]  # 结构化内容（可选）


# 2. 工具返回的结构化内容模型
# 计算表达式结果的数据模型
class CalculationResult(BaseModel):
    """计算表达式结果"""

    # 数学表达式
    expression: str = Field(..., description="数学表达式")
    # 计算结果
    result: float = Field(..., description="计算结果")
    # 计算步骤列表
    steps: List[str] = Field(..., description="计算步骤")
    # 计算时间
    timestamp: datetime = Field(..., description="计算时间")


# 搜索结果的数据模型
class SearchResult(BaseModel):
    """搜索结果"""

    # 结果标题
    title: str = Field(..., description="结果标题")
    # 结果链接
    url: str = Field(..., description="结果链接")
    # 结果摘要
    snippet: str = Field(..., description="结果摘要")
    # 相关度评分，范围0~1
    score: float = Field(..., ge=0, le=1, description="相关度评分")


# 3. 创建 MCP 服务器
# 实例化FastMCP对象，命名为"Structured"
mcp = FastMCP(name="Structured")


# 注册计算表达式工具
@mcp.tool()
async def calculate_expression(expr: str) -> CalculationResult:
    """计算数学表达式并返回结构化结果"""
    try:
        # 使用eval计算表达式
        result = eval(expr)
        # 构建结构化结果并返回
        return CalculationResult(
            expression=expr,
            result=result,
            steps=[f"计算 {expr} = {result}"],
            timestamp=datetime.now(),
        )
    except Exception as e:
        # 捕获异常并抛出ValueError
        raise ValueError(f"计算失败: {e}")


# 注册搜索工具
@mcp.tool()
async def search_web(query: str, limit: int = 5) -> List[SearchResult]:
    """模拟网络搜索并返回结构化结果"""
    # 初始化结果列表
    results = []
    # 生成最多3条模拟搜索结果
    for i in range(min(limit, 3)):
        results.append(
            SearchResult(
                title=f"搜索结果 {i+1}: {query}",
                url=f"https://example.com/result{i+1}",
                snippet=f"这是关于 {query} 的第 {i+1} 个搜索结果...",
                score=0.9 - i * 0.1,
            )
        )
    # 返回搜索结果列表
    return results


# 4. 客户端测试函数
# 测试结构化工具的异步函数
async def test_structured_tools(session: ClientSession):
    """测试结构化工具"""
    # 打印测试计算工具
    print("=== 测试计算工具 ===")
    try:
        # 调用计算表达式工具
        result = await session.call_tool("calculate_expression", {"expr": "2 + 3 * 4"})
        # 打印可读内容
        print("可读内容:")
        for block in result.content:
            # 判断内容块类型并打印
            if isinstance(block, types.TextContent):
                print(f"  文本: {block.text}")
            elif isinstance(block, types.DataContent):
                print(f"  数据: {block.data} (类型: {block.mimeType})")
        # 打印结构化内容
        print("\n结构化内容:")
        if result.structuredContent:
            print(f"  类型: {type(result.structuredContent)}")
            print(f"  内容: {result.structuredContent}")
            # 如果结构化内容为字典，转换为模型并打印详细信息
            if isinstance(result.structuredContent, dict):
                calc_result = CalculationResult(**result.structuredContent)
                print(f"  表达式: {calc_result.expression}")
                print(f"  结果: {calc_result.result}")
                print(f"  步骤: {calc_result.steps}")
                print(f"  时间: {calc_result.timestamp}")
    except Exception as e:
        # 捕获异常并打印错误信息
        print(f"计算工具测试失败: {e}")

    # 打印测试搜索工具
    print("\n=== 测试搜索工具 ===")
    try:
        # 调用搜索工具
        result = await session.call_tool(
            "search_web", {"query": "Python MCP", "limit": 3}
        )
        # 打印可读内容
        print("可读内容:")
        for block in result.content:
            # 判断内容块类型并打印
            if isinstance(block, types.TextContent):
                print(f"  文本: {block.text}")
            elif isinstance(block, types.DataContent):
                print(f"  数据: {block.data} (类型: {block.mimeType})")
        # 打印结构化内容
        print("\n结构化内容:")
        if result.structuredContent:
            print(f"  类型: {type(result.structuredContent)}")
            print(f"  内容: {result.structuredContent}")
            # 如果结构化内容为列表，遍历并打印每个搜索结果
            if isinstance(result.structuredContent, list):
                for i, item in enumerate(result.structuredContent):
                    if isinstance(item, dict):
                        search_result = SearchResult(**item)
                        print(f"  结果 {i+1}:")
                        print(f"    标题: {search_result.title}")
                        print(f"    链接: {search_result.url}")
                        print(f"    摘要: {search_result.snippet}")
                        print(f"    评分: {search_result.score}")
    except Exception as e:
        # 捕获异常并打印错误信息
        print(f"搜索工具测试失败: {e}")


# 5. 客户端主函数
# 客户端运行主函数
async def run_client():
    """运行客户端测试"""
    # 打印启动信息
    print("启动 MCP 客户端测试...")
    # 构造服务器参数，命令为python，参数为当前文件和"serve"
    server_params = StdioServerParameters(command="python", args=[__file__, "serve"])
    # 使用stdio_client连接到服务器
    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(f"可用工具: {[t.name for t in tools.tools]}")
            # 测试结构化工具
            await test_structured_tools(session)


# 6. 服务器入口
# 服务器运行入口函数
def run_server():
    """启动 MCP 服务器"""
    # 打印启动信息
    print("启动 MCP 服务器...")
    # 以stdio方式运行MCP服务器
    mcp.run(transport="stdio")


# 7. 主函数
# 程序主入口
def main():
    # 导入sys模块
    import sys

    # 判断命令行参数，决定运行模式
    if len(sys.argv) >= 2 and sys.argv[1] == "serve":
        # 服务器模式
        run_server()
    else:
        # 客户端模式
        print("启动 MCP 客户端测试...")
        print("将自动启动服务器进程并测试结构化工具")
        asyncio.run(run_client())


# 判断是否为主模块，若是则执行main函数
if __name__ == "__main__":
    main()