fromagentsimportAgent,FileSearchTool,Runner,WebSearchToolagent=Agent(name="Assistant",tools=[WebSearchTool(),FileSearchTool(max_num_results=3,vector_store_ids=["VECTOR_STORE_ID"],),],)asyncdefmain():result=awaitRunner.run(agent,"Which coffee shop should I go to, taking into account my preferences and the weather today in SF?")print(result.final_output)

importjsonfromtyping_extensionsimportTypedDict,AnyfromagentsimportAgent,FunctionTool,RunContextWrapper,function_toolclassLocation(TypedDict):lat:floatlong:float@function_tool# (1)!asyncdeffetch_weather(location:Location)->str:# (2)!"""Fetch the weather for a given location.    Args:        location: The location to fetch the weather for.    """# In real life, we'd fetch the weather from a weather APIreturn"sunny"@function_tool(name_override="fetch_data")# (3)!defread_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.    """# In real life, we'd read the file from the file systemreturn"<file contents>"agent=Agent(name="Assistant",tools=[fetch_weather,read_file],# (4)!)fortoolinagent.tools:ifisinstance(tool,FunctionTool):print(tool.name)print(tool.description)print(json.dumps(tool.params_json_schema,indent=2))print()

fromtypingimportAnyfrompydanticimportBaseModelfromagentsimportRunContextWrapper,FunctionTooldefdo_some_work(data:str)->str:return"done"classFunctionArgs(BaseModel):username:strage:intasyncdefrun_function(ctx:RunContextWrapper[Any],args:str)->str:parsed=FunctionArgs.model_validate_json(args)returndo_some_work(data=f"{parsed.username} is{parsed.age} years old")tool=FunctionTool(name="process_user",description="Processes extracted user data",params_json_schema=FunctionArgs.model_json_schema(),on_invoke_tool=run_function,)

fromagentsimportAgent,Runnerimportasynciospanish_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",),],)asyncdefmain():result=awaitRunner.run(orchestrator_agent,input="Say 'Hello, how are you?' in Spanish.")print(result.final_output)

@function_toolasyncdefrun_my_agent()->str:"""A tool that runs the agent with custom configs"""agent=Agent(name="My agent",instructions="...")result=awaitRunner.run(agent,input="...",max_turns=5,run_config=...)returnstr(result.final_output)

asyncdefextract_json_payload(run_result:RunResult)->str:# Scan the agent’s outputs in reverse order until we find a JSON-like message from a tool call.foriteminreversed(run_result.new_items):ifisinstance(item,ToolCallOutputItem)anditem.output.strip().startswith("{"):returnitem.output.strip()# Fallback to an empty JSON object if nothing was foundreturn"{}"json_tool=data_agent.as_tool(tool_name="get_data_json",tool_description="Run the data agent and return only its JSON payload",custom_output_extractor=extract_json_payload,)

fromagentsimportAgentToolStreamEventasyncdefhandle_stream(event:AgentToolStreamEvent)->None:# Inspect the underlying StreamEvent along with agent metadata.print(f"[stream]{event['agent']['name']} ::{event['event'].type}")billing_agent_tool=billing_agent.as_tool(tool_name="billing_helper",tool_description="Answer billing questions.",on_stream=handle_stream,# Can be sync or async.)

importasynciofromagentsimportAgent,AgentBase,Runner,RunContextWrapperfrompydanticimportBaseModelclassLanguageContext(BaseModel):language_preference:str="french_spanish"deffrench_enabled(ctx:RunContextWrapper[LanguageContext],agent:AgentBase)->bool:"""Enable French for French+Spanish preference."""returnctx.context.language_preference=="french_spanish"# Create specialized agentsspanish_agent=Agent(name="spanish_agent",instructions="You respond in Spanish. Always reply to the user's question in Spanish.",)french_agent=Agent(name="french_agent",instructions="You respond in French. Always reply to the user's question in French.",)# Create orchestrator with conditional toolsorchestrator=Agent(name="orchestrator",instructions=("You are a multilingual assistant. You use the tools given to you to respond to users. ""You must call ALL available tools to provide responses in different languages. ""You never respond in languages yourself, you always use the provided tools."),tools=[spanish_agent.as_tool(tool_name="respond_spanish",tool_description="Respond to the user's question in Spanish",is_enabled=True,# Always enabled),french_agent.as_tool(tool_name="respond_french",tool_description="Respond to the user's question in French",is_enabled=french_enabled,),],)asyncdefmain():context=RunContextWrapper(LanguageContext(language_preference="french_spanish"))result=awaitRunner.run(orchestrator,"How are you?",context=context.context)print(result.final_output)asyncio.run(main())

fromagentsimportfunction_tool,RunContextWrapperfromtypingimportAnydefmy_custom_error_function(context:RunContextWrapper[Any],error:Exception)->str:"""A custom function to provide a user-friendly error message."""print(f"A tool call failed with the following error:{error}")return"An internal server error occurred. Please try again later."@function_tool(failure_error_function=my_custom_error_function)defget_user_profile(user_id:str)->str:"""Fetches a user profile from a mock API.     This function demonstrates a 'flaky' or failing API call.    """ifuser_id=="user_123":return"User profile for user_123 successfully retrieved."else:raiseValueError(f"Could not retrieve profile for user_id:{user_id}. API returned an error.")