OpenAI Agents SDK #3：你的 docstring 就是工具说明书

@function_tool 到底做了什么

• inspect：提取函数签名（参数名、类型注解、默认值）
• griffe：解析 docstring，拿出工具描述和每个参数的说明
• pydantic：把上面两者转成标准 JSON Schema

import json
from typing_extensions import TypedDict, Any
from agents import Agent, FunctionTool, RunContextWrapper, function_tool

class Location(TypedDict):
    lat: float
    long: float

@function_tool
async def fetch_weather(location: Location) -> str:
    """Fetch the weather for a given location.

Args:
        location: The location to fetch the weather for.
    """
    # 实际场景里这里会调天气 API
    return "sunny"

@function_tool(name_override="fetch_data")
def read_file(ctx: RunContextWrapper[Any], path: str, directory: str | None = None) -> str:
    """Read the contents of a file.

Args:
        path: The path to the file to read.
        directory: The directory to read the file from.
    """
    return ""

agent = Agent(
    name="Assistant",
    tools=[fetch_weather, read_file],
)

for tool in agent.tools:
    if isinstance(tool, FunctionTool):
        print(tool.name)
        print(tool.description)
        print(json.dumps(tool.params_json_schema, indent=2))
        print()

1. 函数可以加 context 作为第一个参数（类型是 RunContextWrapper），SDK 调用时会自动注入运行上下文，这个参数不会出现在 schema 里1
2. name_override 可以给工具起个和函数名不同的名字——read_file 函数会被注册为 fetch_data
3. docstring 格式支持 google、sphinx、numpy 三种风格，SDK 会自动检测，也可以显式指定

Pydantic 参数约束：给 LLM 加护栏

from typing import Annotated
from pydantic import Field
from agents import function_tool

# 写法一：默认值形式
@function_tool
def score_a(score: int = Field(..., ge=0, le=100, description="Score from 0 to 100")) -> str:
    return f"Score recorded: {score}"

# 写法二：Annotated 形式（更推荐，类型提示更清晰）
@function_tool
def score_b(score: Annotated[int, Field(..., ge=0, le=100, description="Score from 0 to 100")]) -> str:
    return f"Score recorded: {score}"

工具超时：别让一个慢 API 拖垮整个 Agent

import asyncio
from agents import Agent, Runner, function_tool

@function_tool(timeout=2.0)
async def slow_lookup(query: str) -> str:
    await asyncio.sleep(10)
    return f"Result for {query}"

agent = Agent(
    name="Timeout demo",
    instructions="Use tools when helpful.",
    tools=[slow_lookup],
)

import asyncio
from agents import Agent, Runner, ToolTimeoutError, function_tool

@function_tool(timeout=1.5, timeout_behavior="raise_exception")
async def slow_tool() -> str:
    await asyncio.sleep(5)
    return "done"

agent = Agent(name="Timeout hard-fail", tools=[slow_tool])
try:
    await Runner.run(agent, "Run the tool")
except ToolTimeoutError as e:
    print(f"{e.tool_name} timed out in {e.timeout_seconds} seconds")

max_turns 保护机制：防止工具调用死循环

"A turn is defined as one AI invocation (including any tool calls that might occur)."

——「一个 turn 是一次 AI 调用，包括该次调用里触发的全部工具调用。」

from agents import (
    Agent,
    RunErrorHandlerInput,
    RunErrorHandlerResult,
    Runner,
)

agent = Agent(name="Assistant", instructions="Be concise.")

def on_max_turns(_data: RunErrorHandlerInput[None]) -> RunErrorHandlerResult:
    return RunErrorHandlerResult(
        final_output="I couldn't finish within the turn limit. Please narrow the request.",
        include_in_history=False,
    )

result = Runner.run_sync(
    agent,
    "Analyze this long transcript",
    max_turns=3,
    error_handlers={"max_turns": on_max_turns},
)
print(result.final_output)

工具并发执行：Agent 一次可以 call 多个工具

import asyncio
from agents import Agent, Runner

spanish_agent = Agent(
    name="Spanish agent",
    instructions="You translate the user's message to Spanish",
)
french_agent = Agent(
    name="French agent",
    instructions="You translate the user's message to French",
)

orchestrator_agent = Agent(
    name="orchestrator_agent",
    instructions=(
        "You are a translation agent. You use the tools given to you to translate."
        "If asked for multiple translations, you call the relevant tools."
    ),
    tools=[
        spanish_agent.as_tool(
            tool_name="translate_to_spanish",
            tool_description="Translate the user's message to Spanish",
        ),
        french_agent.as_tool(
            tool_name="translate_to_french",
            tool_description="Translate the user's message to French",
        ),
    ],
)

async def main():
    result = await Runner.run(orchestrator_agent, input="Say 'Hello, how are you?' in Spanish.")
    print(result.final_output)

Built-in Tools：三种场景，三个选择

from agents import Agent, FileSearchTool, Runner, WebSearchTool

agent = Agent(
    name="Assistant",
    tools=[
        WebSearchTool(),
        FileSearchTool(
            max_num_results=3,
            vector_store_ids=["VECTOR_STORE_ID"],
        ),
    ],
)

async def main():
    result = await Runner.run(
        agent,
        "Which coffee shop should I go to, taking into account my preferences and the weather today in SF?"
    )
    print(result.final_output)

三条落地建议