5051525354555657585960

@dataclassclassToolsToFinalOutputResult:is_final_output:bool"""Whether this is the final output. If False, the LLM will run again and receive the tool call    output.    """final_output:Any|None=None"""The final output. Can be None if `is_final_output` is False, otherwise must match the    `output_type` of the agent.    """

7273747576777879808182

classAgentToolStreamEvent(TypedDict):"""Streaming event emitted when an agent is invoked as a tool."""event:StreamEvent"""The streaming event from the nested agent run."""agent:Agent[Any]"""The nested agent emitting the event."""tool_call:ResponseFunctionToolCall|None"""The originating tool call, if available."""

858687

classStopAtTools(TypedDict):stop_at_tool_names:list[str]"""A list of tool names, any of which will stop the agent from running further."""

90919293949596

classMCPConfig(TypedDict):"""Configuration for MCP servers."""convert_schemas_to_strict:NotRequired[bool]"""If True, we will attempt to convert the MCP schemas to strict-mode schemas. This is a    best-effort conversion, so some schemas may not be convertible. Defaults to False.    """

 99100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152

@dataclassclassAgentBase(Generic[TContext]):"""Base class for `Agent` and `RealtimeAgent`."""name:str"""The name of the agent."""handoff_description:str|None=None"""A description of the agent. This is used when the agent is used as a handoff, so that an    LLM knows what it does and when to invoke it.    """tools:list[Tool]=field(default_factory=list)"""A list of tools that the agent can use."""mcp_servers:list[MCPServer]=field(default_factory=list)"""A list of [Model Context Protocol](https://modelcontextprotocol.io/) servers that    the agent can use. Every time the agent runs, it will include tools from these servers in the    list of available tools.    NOTE: You are expected to manage the lifecycle of these servers. Specifically, you must call    `server.connect()` before passing it to the agent, and `server.cleanup()` when the server is no    longer needed.    """mcp_config:MCPConfig=field(default_factory=lambda:MCPConfig())"""Configuration for MCP servers."""asyncdefget_mcp_tools(self,run_context:RunContextWrapper[TContext])->list[Tool]:"""Fetches the available tools from the MCP servers."""convert_schemas_to_strict=self.mcp_config.get("convert_schemas_to_strict",False)returnawaitMCPUtil.get_all_function_tools(self.mcp_servers,convert_schemas_to_strict,run_context,self)asyncdefget_all_tools(self,run_context:RunContextWrapper[TContext])->list[Tool]:"""All agent tools, including MCP tools and function tools."""mcp_tools=awaitself.get_mcp_tools(run_context)asyncdef_check_tool_enabled(tool:Tool)->bool:ifnotisinstance(tool,FunctionTool):returnTrueattr=tool.is_enabledifisinstance(attr,bool):returnattrres=attr(run_context,self)ifinspect.isawaitable(res):returnbool(awaitres)returnbool(res)results=awaitasyncio.gather(*(_check_tool_enabled(t)fortinself.tools))enabled:list[Tool]=[tfort,okinzip(self.tools,results)ifok]return[*mcp_tools,*enabled]

155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573

@dataclassclassAgent(AgentBase,Generic[TContext]):"""An agent is an AI model configured with instructions, tools, guardrails, handoffs and more.    We strongly recommend passing `instructions`, which is the "system prompt" for the agent. In    addition, you can pass `handoff_description`, which is a human-readable description of the    agent, used when the agent is used inside tools/handoffs.    Agents are generic on the context type. The context is a (mutable) object you create. It is    passed to tool functions, handoffs, guardrails, etc.    See `AgentBase` for base parameters that are shared with `RealtimeAgent`s.    """instructions:(str|Callable[[RunContextWrapper[TContext],Agent[TContext]],MaybeAwaitable[str],]|None)=None"""The instructions for the agent. Will be used as the "system prompt" when this agent is    invoked. Describes what the agent should do, and how it responds.    Can either be a string, or a function that dynamically generates instructions for the agent. If    you provide a function, it will be called with the context and the agent instance. It must    return a string.    """prompt:Prompt|DynamicPromptFunction|None=None"""A prompt object (or a function that returns a Prompt). Prompts allow you to dynamically    configure the instructions, tools and other config for an agent outside of your code. Only    usable with OpenAI models, using the Responses API.    """handoffs:list[Agent[Any]|Handoff[TContext,Any]]=field(default_factory=list)"""Handoffs are sub-agents that the agent can delegate to. You can provide a list of handoffs,    and the agent can choose to delegate to them if relevant. Allows for separation of concerns and    modularity.    """model:str|Model|None=None"""The model implementation to use when invoking the LLM.    By default, if not set, the agent will use the default model configured in    `agents.models.get_default_model()` (currently "gpt-4.1").    """model_settings:ModelSettings=field(default_factory=get_default_model_settings)"""Configures model-specific tuning parameters (e.g. temperature, top_p).    """input_guardrails:list[InputGuardrail[TContext]]=field(default_factory=list)"""A list of checks that run in parallel to the agent's execution, before generating a    response. Runs only if the agent is the first agent in the chain.    """output_guardrails:list[OutputGuardrail[TContext]]=field(default_factory=list)"""A list of checks that run on the final output of the agent, after generating a response.    Runs only if the agent produces a final output.    """output_type:type[Any]|AgentOutputSchemaBase|None=None"""The type of the output object. If not provided, the output will be `str`. In most cases,    you should pass a regular Python type (e.g. a dataclass, Pydantic model, TypedDict, etc).    You can customize this in two ways:    1. If you want non-strict schemas, pass `AgentOutputSchema(MyClass, strict_json_schema=False)`.    2. If you want to use a custom JSON schema (i.e. without using the SDK's automatic schema)       creation, subclass and pass an `AgentOutputSchemaBase` subclass.    """hooks:AgentHooks[TContext]|None=None"""A class that receives callbacks on various lifecycle events for this agent.    """tool_use_behavior:(Literal["run_llm_again","stop_on_first_tool"]|StopAtTools|ToolsToFinalOutputFunction)="run_llm_again""""    This lets you configure how tool use is handled.    - "run_llm_again": The default behavior. Tools are run, and then the LLM receives the results        and gets to respond.    - "stop_on_first_tool": The output from the first tool call is treated as the final result.        In other words, it isn’t sent back to the LLM for further processing but is used directly        as the final output.    - A StopAtTools object: The agent will stop running if any of the tools listed in        `stop_at_tool_names` is called.        The final output will be the output of the first matching tool call.        The LLM does not process the result of the tool call.    - A function: If you pass a function, it will be called with the run context and the list of      tool results. It must return a `ToolsToFinalOutputResult`, which determines whether the tool      calls result in a final output.      NOTE: This configuration is specific to FunctionTools. Hosted tools, such as file search,      web search, etc. are always processed by the LLM.    """reset_tool_choice:bool=True"""Whether to reset the tool choice to the default value after a tool has been called. Defaults    to True. This ensures that the agent doesn't enter an infinite loop of tool usage."""def__post_init__(self):fromtypingimportget_originifnotisinstance(self.name,str):raiseTypeError(f"Agent name must be a string, got{type(self.name).__name__}")ifself.handoff_descriptionisnotNoneandnotisinstance(self.handoff_description,str):raiseTypeError(f"Agent handoff_description must be a string or None, "f"got{type(self.handoff_description).__name__}")ifnotisinstance(self.tools,list):raiseTypeError(f"Agent tools must be a list, got{type(self.tools).__name__}")ifnotisinstance(self.mcp_servers,list):raiseTypeError(f"Agent mcp_servers must be a list, got{type(self.mcp_servers).__name__}")ifnotisinstance(self.mcp_config,dict):raiseTypeError(f"Agent mcp_config must be a dict, got{type(self.mcp_config).__name__}")if(self.instructionsisnotNoneandnotisinstance(self.instructions,str)andnotcallable(self.instructions)):raiseTypeError(f"Agent instructions must be a string, callable, or None, "f"got{type(self.instructions).__name__}")if(self.promptisnotNoneandnotcallable(self.prompt)andnothasattr(self.prompt,"get")):raiseTypeError(f"Agent prompt must be a Prompt, DynamicPromptFunction, or None, "f"got{type(self.prompt).__name__}")ifnotisinstance(self.handoffs,list):raiseTypeError(f"Agent handoffs must be a list, got{type(self.handoffs).__name__}")ifself.modelisnotNoneandnotisinstance(self.model,str):from.models.interfaceimportModelifnotisinstance(self.model,Model):raiseTypeError(f"Agent model must be a string, Model, or None, got{type(self.model).__name__}")ifnotisinstance(self.model_settings,ModelSettings):raiseTypeError(f"Agent model_settings must be a ModelSettings instance, "f"got{type(self.model_settings).__name__}")if(# The user sets a non-default modelself.modelisnotNoneand(# The default model is gpt-5is_gpt_5_default()isTrue# However, the specified model is not a gpt-5 modeland(isinstance(self.model,str)isFalseorgpt_5_reasoning_settings_required(self.model)isFalse# type: ignore)# The model settings are not customized for the specified modelandself.model_settings==get_default_model_settings())):# In this scenario, we should use a generic model settings# because non-gpt-5 models are not compatible with the default gpt-5 model settings.# This is a best-effort attempt to make the agent work with non-gpt-5 models.self.model_settings=ModelSettings()ifnotisinstance(self.input_guardrails,list):raiseTypeError(f"Agent input_guardrails must be a list, got{type(self.input_guardrails).__name__}")ifnotisinstance(self.output_guardrails,list):raiseTypeError(f"Agent output_guardrails must be a list, "f"got{type(self.output_guardrails).__name__}")ifself.output_typeisnotNone:from.agent_outputimportAgentOutputSchemaBaseifnot(isinstance(self.output_type,(type,AgentOutputSchemaBase))orget_origin(self.output_type)isnotNone):raiseTypeError(f"Agent output_type must be a type, AgentOutputSchemaBase, or None, "f"got{type(self.output_type).__name__}")ifself.hooksisnotNone:from.lifecycleimportAgentHooksBaseifnotisinstance(self.hooks,AgentHooksBase):raiseTypeError(f"Agent hooks must be an AgentHooks instance or None, "f"got{type(self.hooks).__name__}")if(not(isinstance(self.tool_use_behavior,str)andself.tool_use_behaviorin["run_llm_again","stop_on_first_tool"])andnotisinstance(self.tool_use_behavior,dict)andnotcallable(self.tool_use_behavior)):raiseTypeError(f"Agent tool_use_behavior must be 'run_llm_again', 'stop_on_first_tool', "f"StopAtTools dict, or callable, got{type(self.tool_use_behavior).__name__}")ifnotisinstance(self.reset_tool_choice,bool):raiseTypeError(f"Agent reset_tool_choice must be a boolean, "f"got{type(self.reset_tool_choice).__name__}")defclone(self,**kwargs:Any)->Agent[TContext]:"""Make a copy of the agent, with the given arguments changed.        Notes:            - Uses `dataclasses.replace`, which performs a **shallow copy**.            - Mutable attributes like `tools` and `handoffs` are shallow-copied:              new list objects are created only if overridden, but their contents              (tool functions and handoff objects) are shared with the original.            - To modify these independently, pass new lists when calling `clone()`.        Example:            ```python            new_agent = agent.clone(instructions="New instructions")            ```        """returndataclasses.replace(self,**kwargs)defas_tool(self,tool_name:str|None,tool_description:str|None,custom_output_extractor:(Callable[[RunResult|RunResultStreaming],Awaitable[str]]|None)=None,is_enabled:bool|Callable[[RunContextWrapper[Any],AgentBase[Any]],MaybeAwaitable[bool]]=True,on_stream:Callable[[AgentToolStreamEvent],MaybeAwaitable[None]]|None=None,run_config:RunConfig|None=None,max_turns:int|None=None,hooks:RunHooks[TContext]|None=None,previous_response_id:str|None=None,conversation_id:str|None=None,session:Session|None=None,failure_error_function:ToolErrorFunction|None=default_tool_error_function,)->Tool:"""Transform this agent into a tool, callable by other agents.        This is different from handoffs in two ways:        1. In handoffs, the new agent receives the conversation history. In this tool, the new agent           receives generated input.        2. In handoffs, the new agent takes over the conversation. In this tool, the new agent is           called as a tool, and the conversation is continued by the original agent.        Args:            tool_name: The name of the tool. If not provided, the agent's name will be used.            tool_description: The description of the tool, which should indicate what it does and                when to use it.            custom_output_extractor: A function that extracts the output from the agent. If not                provided, the last message from the agent will be used.            is_enabled: Whether the tool is enabled. Can be a bool or a callable that takes the run                context and agent and returns whether the tool is enabled. Disabled tools are hidden                from the LLM at runtime.            on_stream: Optional callback (sync or async) to receive streaming events from the nested                agent run. The callback receives an `AgentToolStreamEvent` containing the nested                agent, the originating tool call (when available), and each stream event. When                provided, the nested agent is executed in streaming mode.            failure_error_function: If provided, generate an error message when the tool (agent) run                fails. The message is sent to the LLM. If None, the exception is raised instead.        """@function_tool(name_override=tool_nameor_transforms.transform_string_function_style(self.name),description_override=tool_descriptionor"",is_enabled=is_enabled,failure_error_function=failure_error_function,)asyncdefrun_agent(context:ToolContext,input:str)->Any:from.runimportDEFAULT_MAX_TURNS,Runnerresolved_max_turns=max_turnsifmax_turnsisnotNoneelseDEFAULT_MAX_TURNSrun_result:RunResult|RunResultStreamingifon_streamisnotNone:run_result=Runner.run_streamed(starting_agent=self,input=input,context=context.context,run_config=run_config,max_turns=resolved_max_turns,hooks=hooks,previous_response_id=previous_response_id,conversation_id=conversation_id,session=session,)# Dispatch callbacks in the background so slow handlers do not block# event consumption.event_queue:asyncio.Queue[AgentToolStreamEvent|None]=asyncio.Queue()asyncdef_run_handler(payload:AgentToolStreamEvent)->None:"""Execute the user callback while capturing exceptions."""try:maybe_result=on_stream(payload)ifinspect.isawaitable(maybe_result):awaitmaybe_resultexceptException:logger.exception("Error while handling on_stream event for agent tool%s.",self.name,)asyncdefdispatch_stream_events()->None:whileTrue:payload=awaitevent_queue.get()is_sentinel=payloadisNone# None marks the end of the stream.try:ifpayloadisnotNone:await_run_handler(payload)finally:event_queue.task_done()ifis_sentinel:breakdispatch_task=asyncio.create_task(dispatch_stream_events())try:from.stream_eventsimportAgentUpdatedStreamEventcurrent_agent=run_result.current_agentasyncforeventinrun_result.stream_events():ifisinstance(event,AgentUpdatedStreamEvent):current_agent=event.new_agentpayload:AgentToolStreamEvent={"event":event,"agent":current_agent,"tool_call":context.tool_call,}awaitevent_queue.put(payload)finally:awaitevent_queue.put(None)awaitevent_queue.join()awaitdispatch_taskelse:run_result=awaitRunner.run(starting_agent=self,input=input,context=context.context,run_config=run_config,max_turns=resolved_max_turns,hooks=hooks,previous_response_id=previous_response_id,conversation_id=conversation_id,session=session,)ifcustom_output_extractor:returnawaitcustom_output_extractor(run_result)returnrun_result.final_outputreturnrun_agentasyncdefget_system_prompt(self,run_context:RunContextWrapper[TContext])->str|None:ifisinstance(self.instructions,str):returnself.instructionselifcallable(self.instructions):# Inspect the signature of the instructions functionsig=inspect.signature(self.instructions)params=list(sig.parameters.values())# Enforce exactly 2 parametersiflen(params)!=2:raiseTypeError(f"'instructions' callable must accept exactly 2 arguments (context, agent), "f"but got{len(params)}:{[p.nameforpinparams]}")# Call the instructions function properlyifinspect.iscoroutinefunction(self.instructions):returnawaitcast(Awaitable[str],self.instructions(run_context,self))else:returncast(str,self.instructions(run_context,self))elifself.instructionsisnotNone:logger.error(f"Instructions must be a string or a callable function, "f"got{type(self.instructions).__name__}")returnNoneasyncdefget_prompt(self,run_context:RunContextWrapper[TContext])->ResponsePromptParam|None:"""Get the prompt for the agent."""returnawaitPromptUtil.to_model_input(self.prompt,run_context,self)