Claude Code 学习站

Agent SDK 结构化输出(Structured Outputs)参考

介绍如何用 JSON Schema、Zod 或 Pydantic 让 Claude Agent SDK 在多轮工具调用后返回经校验的类型化 JSON 数据。

本页目录8
AI 摘要 · 已核查整理于 2026-08-02原文:Get structured output from agents(Anthropic)Agent SDK结构化输出JSON SchemaTypeScript/Python
要点速览
  • 通过 `query()` 的 `outputFormat`(TS)/ `output_format`(Python)选项传入 `{ type: "json_schema", schema }`,结果消息的 `structured_output` 字段即为校验后的数据
  • 推荐用 Zod(TS)或 Pydantic(Python)生成 schema 并做强类型解析;SDK 按 JSON Schema draft-07 校验,Zod 需显式传 `target: "draft-7"`
  • schema 不合法会在启动时直接报错终止(v2.1.205 起);`format` 关键字仅作为注解,不会被强制校验
  • 校验失败会自动重试,超过重试次数后结果 `subtype` 为 `error_max_structured_output_retries`;模型 fallback 也可能在无重试补位时导致同样的失败
  • 即使 `subtype` 为 `success`,也可能没有 `structured_output`(例如任务未产出结构化结果),需一并当作失败处理
  • agent 在产出结构化输出前仍可自由使用任意工具(Grep、Bash 等)完成多步任务

本文是对 Claude Agent SDK 官方文档某页的中文整理,完整与最新内容以原文为准:https://code.claude.com/docs/en/agent-sdk/structured-outputs

概述

结构化输出(Structured Outputs)允许你定义希望从 agent 拿回的数据的确切形状。agent 仍可自主使用任意工具完成任务,但最终会返回与你的 schema 匹配、经过校验的 JSON。

做法:定义一个 JSON Schema 描述所需结构,SDK 会据此校验输出,遇到不匹配时自动重新提示(re-prompt)agent。如果在重试次数上限内仍未通过校验,结果将是一个错误而非结构化数据(见下文「错误处理」)。

若需要完整的类型安全,可使用 Zod(TypeScript)或 Pydantic(Python)来定义 schema,从而拿到强类型对象。

为什么需要结构化输出

agent 默认返回自由格式文本,适合聊天场景,但不适合程序化使用。结构化输出能给你直接传给应用逻辑、数据库或 UI 组件的类型化数据。

例如一个食谱应用中,agent 联网搜索并返回食谱:没有结构化输出时,你拿到的是需要自行解析的自由文本(标题、耗时字符串、原料与步骤混杂、格式还可能不一致);有了结构化输出后,你直接定义想要的形状,拿到可在应用中直接使用的类型化数据,例如:

{
  "name": "Chocolate Chip Cookies",
  "prep_time_minutes": 15,
  "cook_time_minutes": 10,
  "ingredients": [
    { "item": "all-purpose flour", "amount": 2.25, "unit": "cups" },
    { "item": "butter, softened", "amount": 1, "unit": "cup" }
  ],
  "steps": ["Preheat oven to 375°F", "Cream butter and sugar"]
}

快速开始

定义一个 JSON Schema 描述想要的数据形状,然后通过 query()outputFormat(TypeScript)或 output_format(Python)选项传入。agent 完成后,结果消息(result message)中会包含 structured_output 字段,即符合 schema 的校验后数据。

下例让 agent 调研 Anthropic 公司,并以结构化输出返回公司名称、成立年份和总部地点。

import { query } from "@anthropic-ai/claude-agent-sdk";

// Define the shape of data you want back
const schema = {
  type: "object",
  properties: {
    company_name: { type: "string" },
    founded_year: { type: "number" },
    headquarters: { type: "string" }
  },
  required: ["company_name"]
};

try {
  for await (const message of query({
    prompt: "Research Anthropic and provide key company information",
    options: {
      outputFormat: {
        type: "json_schema",
        schema: schema
      }
    }
  })) {
    // The result message contains structured_output with validated data
    if (message.type === "result" && message.subtype === "success" && message.structured_output) {
      console.log(message.structured_output);
      // { company_name: "Anthropic", founded_year: 2021, headquarters: "San Francisco, CA" }
    }
  }
} catch (error) {
  // A single-shot query() throws after yielding an error result, such as
  // error_max_structured_output_retries; see the Error handling section.
  console.error(`Session ended with an error: ${error}`);
}
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

# Define the shape of data you want back
schema = {
    "type": "object",
    "properties": {
        "company_name": {"type": "string"},
        "founded_year": {"type": "number"},
        "headquarters": {"type": "string"},
    },
    "required": ["company_name"],
}


async def main():
    try:
        async for message in query(
            prompt="Research Anthropic and provide key company information",
            options=ClaudeAgentOptions(
                output_format={"type": "json_schema", "schema": schema}
            ),
        ):
            # The result message contains structured_output with validated data
            if isinstance(message, ResultMessage) and message.structured_output:
                print(message.structured_output)
                # {'company_name': 'Anthropic', 'founded_year': 2021, 'headquarters': 'San Francisco, CA'}
    except Exception as error:
        # A single-shot query() raises after yielding an error result, such as
        # error_max_structured_output_retries; see the Error handling section.
        print(f"Session ended with an error: {error}")


asyncio.run(main())

类型安全的 Schema:Zod 与 Pydantic

除了手写 JSON Schema,也可以用 Zod(TypeScript)或 Pydantic(Python)定义 schema。这两个库会为你生成 JSON Schema,并让你把响应解析为在整个代码库中都能获得自动补全和类型检查的强类型对象。

下例定义了一个「功能实现计划」的 schema,包含摘要、步骤列表(每步含复杂度)和潜在风险。agent 规划该功能后返回一个类型化的 FeaturePlan 对象,你可以直接访问 plan.summary、遍历 plan.steps,并获得完整的类型安全。

SDK 按 JSON Schema draft-07 校验 schema,声明了更新版本的 schema 会被拒绝。Zod 默认生成 draft 2020-12 目标,因此转换时需传入 target: "draft-7"

import { z } from "zod";
import { query } from "@anthropic-ai/claude-agent-sdk";

// Define schema with Zod
const FeaturePlan = z.object({
  feature_name: z.string(),
  summary: z.string(),
  steps: z.array(
    z.object({
      step_number: z.number(),
      description: z.string(),
      estimated_complexity: z.enum(["low", "medium", "high"])
    })
  ),
  risks: z.array(z.string())
});

type FeaturePlan = z.infer<typeof FeaturePlan>;

// Convert to JSON Schema using the draft-07 target the SDK expects
const schema = z.toJSONSchema(FeaturePlan, { target: "draft-7" });

// Use in query
try {
  for await (const message of query({
    prompt:
      "Plan how to add dark mode support to a React app. Break it into implementation steps.",
    options: {
      outputFormat: {
        type: "json_schema",
        schema: schema
      }
    }
  })) {
    if (message.type === "result" && message.subtype === "success" && message.structured_output) {
      // Validate and get fully typed result
      const parsed = FeaturePlan.safeParse(message.structured_output);
      if (parsed.success) {
        const plan: FeaturePlan = parsed.data;
        console.log(`Feature: ${plan.feature_name}`);
        console.log(`Summary: ${plan.summary}`);
        plan.steps.forEach((step) => {
          console.log(`${step.step_number}. [${step.estimated_complexity}] ${step.description}`);
        });
      }
    }
  }
} catch (error) {
  // A single-shot query() throws after yielding an error result, such as
  // error_max_structured_output_retries; see the Error handling section.
  console.error(`Session ended with an error: ${error}`);
}
import asyncio
from pydantic import BaseModel
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage


class Step(BaseModel):
    step_number: int
    description: str
    estimated_complexity: str  # 'low', 'medium', 'high'


class FeaturePlan(BaseModel):
    feature_name: str
    summary: str
    steps: list[Step]
    risks: list[str]


async def main():
    try:
        async for message in query(
            prompt="Plan how to add dark mode support to a React app. Break it into implementation steps.",
            options=ClaudeAgentOptions(
                output_format={
                    "type": "json_schema",
                    "schema": FeaturePlan.model_json_schema(),
                }
            ),
        ):
            if isinstance(message, ResultMessage) and message.structured_output:
                # Validate and get fully typed result
                plan = FeaturePlan.model_validate(message.structured_output)
                print(f"Feature: {plan.feature_name}")
                print(f"Summary: {plan.summary}")
                for step in plan.steps:
                    print(
                        f"{step.step_number}. [{step.estimated_complexity}] {step.description}"
                    )
    except Exception as error:
        # A single-shot query() raises after yielding an error result, such as
        # error_max_structured_output_retries; see the Error handling section.
        print(f"Session ended with an error: {error}")


asyncio.run(main())

优势:

  • 完整的类型推断(TypeScript)与类型提示(Python)
  • 通过 safeParse()model_validate() 做运行时校验
  • 更清晰的错误信息
  • schema 可组合、可复用

输出格式配置(Output format configuration)

outputFormat(TypeScript)或 output_format(Python)选项接受一个包含以下字段的对象:

字段类型默认值说明
type"json_schema"固定设为 "json_schema" 以启用结构化输出
schemaJSON Schema 对象定义输出结构的 JSON Schema 对象。可用 z.toJSONSchema(schema, { target: "draft-7" })(Zod)或 .model_json_schema()(Pydantic)生成

SDK 支持标准 JSON Schema 特性,包括所有基础类型(object、array、string、number、boolean、null)、enumconstrequired、嵌套对象以及 $ref 定义。完整的支持特性与限制列表见 JSON Schema limitations

不合法的 JSON Schema 会在运行启动时直接失败,并报出具体问题所在(v2.1.205 之前,不合法的 schema 会被静默忽略,agent 转而返回非结构化文本)。

format 关键字(例如 "format": "email" )会被当作注解接受,但不会被 SDK 的校验器强制执行(v2.1.205 之前,任何包含 format 的 schema 都会被视为不合法)。

示例:TODO 追踪 agent

此示例展示结构化输出如何与多步工具调用配合使用。agent 需要在代码库中查找 TODO 注释,再为每一条查询 git blame 信息。它会自主决定使用哪些工具(Grep 搜索、Bash 运行 git 命令),并将结果合并为单个结构化响应。

schema 中的 authordate 是可选字段,因为并非所有文件都能取到 git blame 信息;agent 会填入能找到的内容,其余省略。

import { query } from "@anthropic-ai/claude-agent-sdk";

// Define structure for TODO extraction
const todoSchema = {
  type: "object",
  properties: {
    todos: {
      type: "array",
      items: {
        type: "object",
        properties: {
          text: { type: "string" },
          file: { type: "string" },
          line: { type: "number" },
          author: { type: "string" },
          date: { type: "string" }
        },
        required: ["text", "file", "line"]
      }
    },
    total_count: { type: "number" }
  },
  required: ["todos", "total_count"]
};

// Agent uses Grep to find TODOs, Bash to get git blame info
try {
  for await (const message of query({
    prompt: "Find all TODO comments in this codebase and identify who added them",
    options: {
      outputFormat: {
        type: "json_schema",
        schema: todoSchema
      }
    }
  })) {
    if (message.type === "result" && message.subtype === "success" && message.structured_output) {
      const data = message.structured_output as { total_count: number; todos: Array<{ file: string; line: number; text: string; author?: string; date?: string }> };
      console.log(`Found ${data.total_count} TODOs`);
      data.todos.forEach((todo) => {
        console.log(`${todo.file}:${todo.line} - ${todo.text}`);
        if (todo.author) {
          console.log(`  Added by ${todo.author} on ${todo.date}`);
        }
      });
    }
  }
} catch (error) {
  // A single-shot query() throws after yielding an error result, such as
  // error_max_structured_output_retries; see the Error handling section.
  console.error(`Session ended with an error: ${error}`);
}
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

# Define structure for TODO extraction
todo_schema = {
    "type": "object",
    "properties": {
        "todos": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "text": {"type": "string"},
                    "file": {"type": "string"},
                    "line": {"type": "number"},
                    "author": {"type": "string"},
                    "date": {"type": "string"},
                },
                "required": ["text", "file", "line"],
            },
        },
        "total_count": {"type": "number"},
    },
    "required": ["todos", "total_count"],
}


async def main():
    # Agent uses Grep to find TODOs, Bash to get git blame info
    try:
        async for message in query(
            prompt="Find all TODO comments in this codebase and identify who added them",
            options=ClaudeAgentOptions(
                output_format={"type": "json_schema", "schema": todo_schema}
            ),
        ):
            if isinstance(message, ResultMessage) and message.structured_output:
                data = message.structured_output
                print(f"Found {data['total_count']} TODOs")
                for todo in data["todos"]:
                    print(f"{todo['file']}:{todo['line']} - {todo['text']}")
                    if "author" in todo:
                        print(f"  Added by {todo['author']} on {todo['date']}")
    except Exception as error:
        # A single-shot query() raises after yielding an error result, such as
        # error_max_structured_output_retries; see the Error handling section.
        print(f"Session ended with an error: {error}")


asyncio.run(main())

错误处理(Error handling)

当 agent 无法生成符合 schema 的合法 JSON 时,结构化输出生成会失败。常见原因包括:schema 相对任务过于复杂、任务本身含糊不清、或 agent 在尝试修正校验错误时达到重试上限。此外,即使没有任何校验失败,也可能出错:模型 fallback 可能会在流式生成过程中撤回一个已经完成的输出,若没有重试补上,该次运行会以相同的错误结束。可通过检查结果消息上的 errors 列表来区分这两类原因,再决定是否需要调试 schema。

出错时,结果消息(result message)的 subtype 会指明具体原因:

subtype含义
success输出已生成并成功通过校验
error_max_structured_output_retries多次尝试后仍没有得到合法输出(可能是校验失败,也可能是模型 fallback 撤回输出且无成功重试)

一次运行也可能以 subtype: success 结束,但却没有 structured_output 值——例如运行完成了但 agent 并未产出结构化输出。这种情况也应当被当作失败处理。下例只在 subtypesuccessstructured_output 存在时才视为成功,其余情况一律当作失败处理:

import { query } from "@anthropic-ai/claude-agent-sdk";

const contactSchema = {
  type: "object",
  properties: {
    name: { type: "string" },
    email: { type: "string" }
  },
  required: ["name"]
};

try {
  for await (const msg of query({
    prompt: "Extract contact info from the document",
    options: {
      outputFormat: {
        type: "json_schema",
        schema: contactSchema
      }
    }
  })) {
    if (msg.type === "result") {
      if (msg.subtype === "success" && msg.structured_output) {
        // Use the validated output
        console.log(msg.structured_output);
      } else if (msg.subtype === "error_max_structured_output_retries") {
        console.error("Could not produce valid output");
      } else {
        console.error("Run ended without a structured output");
      }
    }
  }
} catch (error) {
  // A single-shot query() throws after yielding an error result. If the
  // failure was an error result, the error subtype branches above have
  // already run; connection or process failures yield no result message.
  console.log(`Session ended with an error: ${error}`);
}
import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

contact_schema = {
    "type": "object",
    "properties": {
        "name": {"type": "string"},
        "email": {"type": "string"},
    },
    "required": ["name"],
}


async def main():
    try:
        async for message in query(
            prompt="Extract contact info from the document",
            options=ClaudeAgentOptions(
                output_format={"type": "json_schema", "schema": contact_schema}
            ),
        ):
            if isinstance(message, ResultMessage):
                if message.subtype == "success" and message.structured_output:
                    # Use the validated output
                    print(message.structured_output)
                elif message.subtype == "error_max_structured_output_retries":
                    print("Could not produce valid output")
                else:
                    print("Run ended without a structured output")
    except Exception as error:
        # A single-shot query() raises after yielding an error result. If the
        # failure was an error result, the error subtype branches above have
        # already run; connection or process failures yield no result message.
        print(f"Session ended with an error: {error}")


asyncio.run(main())

减少出错的建议:

  • 保持 schema 聚焦:嵌套很深、必填字段很多的 schema 更难满足,建议从简单开始,按需增加复杂度。
  • 让 schema 匹配任务:如果任务不一定能获取到 schema 要求的全部信息,应将这些字段设为可选。
  • 使用清晰的 prompt:含糊的 prompt 会让 agent 更难判断应该产出什么样的输出。

相关资源

  • JSON Schema 文档:学习用于定义嵌套对象、数组、枚举及校验约束等复杂 schema 的 JSON Schema 语法
  • API Structured Outputs:在无工具调用的单轮请求中直接对 Claude API 使用结构化输出
  • Custom tools:在返回结构化输出前,为 agent 提供可在执行过程中调用的自定义工具