跳转至

Agents

ToolsToFinalOutputFunction module-attribute 

ToolsToFinalOutputFunction: TypeAlias = Callable[[RunContextWrapper[TContext], list[FunctionToolResult]], MaybeAwaitable[ToolsToFinalOutputResult]]

一个函数，接收运行上下文和工具结果列表，并返回 ToolsToFinalOutputResult。

ToolsToFinalOutputResult dataclass

Source code in agents/agent.py 
30
31
32
33
34
35
36
@dataclass
class ToolsToFinalOutputResult:
    is_final_output: bool
    """是否为最终输出。如果为 False，LLM 将再次运行并接收工具调用的输出。"""

    final_output: Any | None = None
    """最终输出。如果 `is_final_output` 为 False，可以为 None，否则必须与 agent 的 `output_type` 匹配。"""

is_final_output instance-attribute 

is_final_output: bool

是否为最终输出。如果为 False，LLM 将再次运行并接收工具调用的输出。

final_output class-attribute instance-attribute 

final_output: Any | None = None

最终输出。如果 is_final_output 为 False，可以为 None，否则必须与 agent 的 output_type 匹配。

StopAtTools

Bases: TypedDict

Source code in agents/agent.py 
46
47
48
class StopAtTools(TypedDict):
    stop_at_tool_names: list[str]
    """工具名称列表，调用其中任意一个工具都会让 agent 停止继续运行。"""

stop_at_tool_names instance-attribute 

stop_at_tool_names: list[str]

工具名称列表，调用其中任意一个工具都会让 agent 停止继续运行。

MCPConfig

Bases: TypedDict

MCP 服务器的配置。

Source code in agents/agent.py 
51
52
53
54
55
class MCPConfig(TypedDict):
    """MCP 服务器的配置。"""

    convert_schemas_to_strict: NotRequired[bool]
    """如果为 True，我们会尝试将 MCP schemas 转换为严格模式。这是尽力而为的转换，因此有些 schema 可能无法转换。默认为 False。"""

convert_schemas_to_strict instance-attribute 

convert_schemas_to_strict: NotRequired[bool]

如果为 True，我们会尝试将 MCP schemas 转换为严格模式。这是尽力而为的转换，因此有些 schema 可能无法转换。默认为 False。

Agent dataclass

Bases: Generic[TContext]

Agent 是一个配置了指令、工具、护栏、交接等的 AI 模型。

强烈建议传递 instructions，它是 agent 的“系统提示词”。此外，你可以传递 handoff_description，这是 agent 的可读描述，在 agent 作为工具/交接时使用。

Agent 对上下文类型是泛型的。上下文是你创建的（可变的）对象，会传递给工具函数、交接、护栏等。

Source code in agents/agent.py 
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
@dataclass
class Agent(Generic[TContext]):
    """Agent 是一个配置了指令、工具、护栏、交接等的 AI 模型。

    强烈建议传递 `instructions`，它是 agent 的“系统提示词”。此外，你可以传递 `handoff_description`，这是 agent 的可读描述，在 agent 作为工具/交接时使用。

    Agent 对上下文类型是泛型的。上下文是你创建的（可变的）对象，会传递给工具函数、交接、护栏等。
    """

    name: str
    """agent 的名称。"""

    instructions: (
        str
        | Callable[
            [RunContextWrapper[TContext], Agent[TContext]],
            MaybeAwaitable[str],
        ]
        | None
    ) = None
    """agent 的指令。在调用该 agent 时会作为“系统提示词”使用，描述 agent 应该做什么以及如何响应。

    可以是字符串，也可以是动态生成指令的函数。如果你提供函数，会用上下文和 agent 实例调用它，必须返回字符串。
    """

    handoff_description: str | None = None
    """agent 的描述。当 agent 作为交接使用时，LLM 会用它来了解 agent 的作用以及何时调用。"""

    handoffs: list[Agent[Any] | Handoff[TContext]] = field(default_factory=list)
    """交接是 agent 可以委托的子 agent。你可以提供交接列表，agent 可以根据需要选择委托给它们。这样有助于关注点分离和模块化。"""

    model: str | Model | None = None
    """调用 LLM 时使用的模型实现。

    默认情况下，如果未设置，agent 会使用 `openai_provider.DEFAULT_MODEL`（当前为 "gpt-4o"）配置的默认模型。
    """

    model_settings: ModelSettings = field(default_factory=ModelSettings)
    """配置模型相关的参数（如 temperature、top_p 等）。"""

    tools: list[Tool] = field(default_factory=list)
    """agent 可用的工具列表。"""

    mcp_servers: list[MCPServer] = field(default_factory=list)
    """agent 可用的 [Model Context Protocol](https://modelcontextprotocol.io/) 服务器列表。每次 agent 运行时，会将这些服务器的工具加入可用工具列表。

    注意：你需要自己管理这些服务器的生命周期。具体来说，必须在传递给 agent 前调用 `server.connect()`，不再需要时调用 `server.cleanup()`。
    """

    mcp_config: MCPConfig = field(default_factory=lambda: MCPConfig())
    """MCP 服务器的配置。"""

    input_guardrails: list[InputGuardrail[TContext]] = field(default_factory=list)
    """在 agent 执行前并行运行的检查列表，仅在 agent 是链中的第一个 agent 时运行。"""

    output_guardrails: list[OutputGuardrail[TContext]] = field(default_factory=list)
    """在 agent 生成最终输出后运行的检查列表，仅在 agent 产生最终输出时运行。"""

    output_type: type[Any] | AgentOutputSchemaBase | None = None
    """输出对象的类型。如果未提供，输出为 `str`。大多数情况下，你应该传递常规 Python 类型（如 dataclass、Pydantic model、TypedDict 等）。
    你可以通过两种方式自定义：
    1. 如果需要非严格 schema，传递 `AgentOutputSchema(MyClass, strict_json_schema=False)`。
    2. 如果需要自定义 JSON schema（即不使用 SDK 自动 schema），继承并传递 `AgentOutputSchemaBase` 子类。
    """

    hooks: AgentHooks[TContext] | None = None
    """用于接收 agent 各生命周期事件回调的类。"""

    tool_use_behavior: (
        Literal["run_llm_again", "stop_on_first_tool"] | StopAtTools | ToolsToFinalOutputFunction
    ) = "run_llm_again"
    """用于配置工具使用的行为。
    - "run_llm_again"：默认行为。工具运行后，LLM 会收到结果并响应。
    - "stop_on_first_tool"：第一个工具调用的输出作为最终输出，LLM 不处理工具调用结果。
    - 工具名称列表：如果调用了列表中的任意工具，agent 会停止运行，最终输出为第一个匹配工具的输出，LLM 不处理工具调用结果。
    - 函数：如果传递函数，会用运行上下文和工具结果列表调用，必须返回 `ToolToFinalOutputResult`，决定工具调用是否为最终输出。

      注意：此配置仅适用于 FunctionTools。托管工具（如文件搜索、网页搜索等）始终由 LLM 处理。
    """

    reset_tool_choice: bool = True
    """工具调用后是否将工具选择重置为默认值。默认为 True。这样可以防止 agent 陷入工具调用的无限循环。"""

    def clone(self, **kwargs: Any) -> Agent[TContext]:
        """复制 agent，并修改指定参数。例如：
        ```
        new_agent = agent.clone(instructions="新的指令")
        ```
        """
        return dataclasses.replace(self, **kwargs)

    def as_tool(
        self,
        tool_name: str | None,
        tool_description: str | None,
        custom_output_extractor: Callable[[RunResult], Awaitable[str]] | None = None,
    ) -> Tool:
        """将该 agent 转换为工具，可被其他 agent 调用。

        与交接有两点不同：
        1. 在交接中，新 agent 会接收对话历史；在此工具中，新 agent 接收生成的输入。
        2. 在交接中，新 agent 接管对话；在此工具中，新 agent 作为工具被调用，对话由原 agent 继续。

        参数:
            tool_name: 工具名称。如果未提供，则使用 agent 的名称。
            tool_description: 工具描述，应说明其作用及使用时机。
            custom_output_extractor: 从 agent 提取输出的函数。如果未提供，则使用 agent 的最后一条消息。
        """

        @function_tool(
            name_override=tool_name or _transforms.transform_string_function_style(self.name),
            description_override=tool_description or "",
        )
        async def run_agent(context: RunContextWrapper, input: str) -> str:
            from .run import Runner

            output = await Runner.run(
                starting_agent=self,
                input=input,
                context=context.context,
            )
            if custom_output_extractor:
                return await custom_output_extractor(output)

            return ItemHelpers.text_message_outputs(output.new_items)

        return run_agent

    async def get_system_prompt(self, run_context: RunContextWrapper[TContext]) -> str | None:
        """获取 agent 的系统提示词。"""
        if isinstance(self.instructions, str):
            return self.instructions
        elif callable(self.instructions):
            if inspect.iscoroutinefunction(self.instructions):
                return await cast(Awaitable[str], self.instructions(run_context, self))
            else:
                return cast(str, self.instructions(run_context, self))
        elif self.instructions is not None:
            logger.error(f"Instructions must be a string or a function, got {self.instructions}")

        return None

    async def get_mcp_tools(self) -> list[Tool]:
        """从 MCP 服务器获取可用工具。"""
        convert_schemas_to_strict = self.mcp_config.get("convert_schemas_to_strict", False)
        return await MCPUtil.get_all_function_tools(self.mcp_servers, convert_schemas_to_strict)

    async def get_all_tools(self) -> list[Tool]:
        """所有 agent 工具，包括 MCP 工具和函数工具。"""
        mcp_tools = await self.get_mcp_tools()
        return mcp_tools + self.tools

name instance-attribute 

name: str

agent 的名称。

instructions class-attribute instance-attribute 

instructions: str | Callable[[RunContextWrapper[TContext], Agent[TContext]], MaybeAwaitable[str]] | None = None

agent 的指令。在调用该 agent 时会作为“系统提示词”使用，描述 agent 应该做什么以及如何响应。

可以是字符串，也可以是动态生成指令的函数。如果你提供函数，会用上下文和 agent 实例调用它，必须返回字符串。

handoff_description class-attribute instance-attribute 

handoff_description: str | None = None

agent 的描述。当 agent 作为交接使用时，LLM 会用它来了解 agent 的作用以及何时调用。

handoffs class-attribute instance-attribute 

handoffs: list[Agent[Any] | Handoff[TContext]] = field(default_factory=list)

交接是 agent 可以委托的子 agent。你可以提供交接列表，agent 可以根据需要选择委托给它们。这样有助于关注点分离和模块化。

model class-attribute instance-attribute 

model: str | Model | None = None

调用 LLM 时使用的模型实现。

默认情况下，如果未设置，agent 会使用 openai_provider.DEFAULT_MODEL（当前为 "gpt-4o"）配置的默认模型。

model_settings class-attribute instance-attribute 

model_settings: ModelSettings = field(default_factory=ModelSettings)

配置模型相关的参数（如 temperature、top_p 等）。

tools class-attribute instance-attribute 

tools: list[Tool] = field(default_factory=list)

agent 可用的工具列表。

mcp_servers class-attribute instance-attribute 

mcp_servers: list[MCPServer] = field(default_factory=list)

agent 可用的 Model Context Protocol 服务器列表。每次 agent 运行时，会将这些服务器的工具加入可用工具列表。

注意：你需要自己管理这些服务器的生命周期。具体来说，必须在传递给 agent 前调用 server.connect()，不再需要时调用 server.cleanup()。

mcp_config class-attribute instance-attribute 

mcp_config: MCPConfig = field(default_factory=lambda: MCPConfig())

MCP 服务器的配置。

input_guardrails class-attribute instance-attribute 

input_guardrails: list[InputGuardrail[TContext]] = field(default_factory=list)

在 agent 执行前并行运行的检查列表，仅在 agent 是链中的第一个 agent 时运行。

output_guardrails class-attribute instance-attribute 

output_guardrails: list[OutputGuardrail[TContext]] = field(default_factory=list)

在 agent 生成最终输出后运行的检查列表，仅在 agent 产生最终输出时运行。

output_type class-attribute instance-attribute 

output_type: type[Any] | AgentOutputSchemaBase | None = None

输出对象的类型。如果未提供，输出为 str。大多数情况下，你应该传递常规 Python 类型（如 dataclass、Pydantic model、TypedDict 等）。 你可以通过两种方式自定义： 1. 如果需要非严格 schema，传递 AgentOutputSchema(MyClass, strict_json_schema=False)。 2. 如果需要自定义 JSON schema（即不使用 SDK 自动 schema），继承并传递 AgentOutputSchemaBase 子类。

hooks class-attribute instance-attribute 

hooks: AgentHooks[TContext] | None = None

用于接收 agent 各生命周期事件回调的类。

tool_use_behavior class-attribute instance-attribute 

tool_use_behavior: Literal['run_llm_again', 'stop_on_first_tool'] | StopAtTools | ToolsToFinalOutputFunction = 'run_llm_again'

用于配置工具使用的行为。 - "run_llm_again"：默认行为。工具运行后，LLM 会收到结果并响应。 - "stop_on_first_tool"：第一个工具调用的输出作为最终输出，LLM 不处理工具调用结果。 - 工具名称列表：如果调用了列表中的任意工具，agent 会停止运行，最终输出为第一个匹配工具的输出，LLM 不处理工具调用结果。 - 函数：如果传递函数，会用运行上下文和工具结果列表调用，必须返回 ToolToFinalOutputResult，决定工具调用是否为最终输出。

注意：此配置仅适用于 FunctionTools。托管工具（如文件搜索、网页搜索等）始终由 LLM 处理。

reset_tool_choice class-attribute instance-attribute 

reset_tool_choice: bool = True

工具调用后是否将工具选择重置为默认值。默认为 True。这样可以防止 agent 陷入工具调用的无限循环。

clone 

clone(**kwargs: Any) -> Agent[TContext]

复制 agent，并修改指定参数。例如： 

new_agent = agent.clone(instructions="新的指令")

Source code in agents/agent.py 
141
142
143
144
145
146
147
def clone(self, **kwargs: Any) -> Agent[TContext]:
    """复制 agent，并修改指定参数。例如：
    ```
    new_agent = agent.clone(instructions="新的指令")
    ```
    """
    return dataclasses.replace(self, **kwargs)

as_tool 

as_tool(tool_name: str | None, tool_description: str | None, custom_output_extractor: Callable[[RunResult], Awaitable[str]] | None = None) -> Tool

将该 agent 转换为工具，可被其他 agent 调用。

与交接有两点不同： 1. 在交接中，新 agent 会接收对话历史；在此工具中，新 agent 接收生成的输入。 2. 在交接中，新 agent 接管对话；在此工具中，新 agent 作为工具被调用，对话由原 agent 继续。

参数

tool_name: 工具名称。如果未提供，则使用 agent 的名称。 tool_description: 工具描述，应说明其作用及使用时机。 custom_output_extractor: 从 agent 提取输出的函数。如果未提供，则使用 agent 的最后一条消息。

Source code in agents/agent.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
def as_tool(
    self,
    tool_name: str | None,
    tool_description: str | None,
    custom_output_extractor: Callable[[RunResult], Awaitable[str]] | None = None,
) -> Tool:
    """将该 agent 转换为工具，可被其他 agent 调用。

    与交接有两点不同：
    1. 在交接中，新 agent 会接收对话历史；在此工具中，新 agent 接收生成的输入。
    2. 在交接中，新 agent 接管对话；在此工具中，新 agent 作为工具被调用，对话由原 agent 继续。

    参数:
        tool_name: 工具名称。如果未提供，则使用 agent 的名称。
        tool_description: 工具描述，应说明其作用及使用时机。
        custom_output_extractor: 从 agent 提取输出的函数。如果未提供，则使用 agent 的最后一条消息。
    """

    @function_tool(
        name_override=tool_name or _transforms.transform_string_function_style(self.name),
        description_override=tool_description or "",
    )
    async def run_agent(context: RunContextWrapper, input: str) -> str:
        from .run import Runner

        output = await Runner.run(
            starting_agent=self,
            input=input,
            context=context.context,
        )
        if custom_output_extractor:
            return await custom_output_extractor(output)

        return ItemHelpers.text_message_outputs(output.new_items)

    return run_agent

get_system_prompt async 

get_system_prompt(run_context: RunContextWrapper[TContext]) -> str | None

获取 agent 的系统提示词。

Source code in agents/agent.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
async def get_system_prompt(self, run_context: RunContextWrapper[TContext]) -> str | None:
    """获取 agent 的系统提示词。"""
    if isinstance(self.instructions, str):
        return self.instructions
    elif callable(self.instructions):
        if inspect.iscoroutinefunction(self.instructions):
            return await cast(Awaitable[str], self.instructions(run_context, self))
        else:
            return cast(str, self.instructions(run_context, self))
    elif self.instructions is not None:
        logger.error(f"Instructions must be a string or a function, got {self.instructions}")

    return None

get_mcp_tools async 

get_mcp_tools() -> list[Tool]

从 MCP 服务器获取可用工具。

Source code in agents/agent.py 
200
201
202
203
async def get_mcp_tools(self) -> list[Tool]:
    """从 MCP 服务器获取可用工具。"""
    convert_schemas_to_strict = self.mcp_config.get("convert_schemas_to_strict", False)
    return await MCPUtil.get_all_function_tools(self.mcp_servers, convert_schemas_to_strict)

get_all_tools async 

get_all_tools() -> list[Tool]

所有 agent 工具，包括 MCP 工具和函数工具。

Source code in agents/agent.py 
205
206
207
208
async def get_all_tools(self) -> list[Tool]:
    """所有 agent 工具，包括 MCP 工具和函数工具。"""
    mcp_tools = await self.get_mcp_tools()
    return mcp_tools + self.tools