### Full example Source: https://openai.github.io/openai-agents-python/voice/quickstart A complete example combining agent setup, voice pipeline initialization, and execution. ```python import asyncio import random import numpy as np import sounddevice as sd from agents import ( Agent, function_tool, set_tracing_disabled, ) from agents.voice import ( AudioInput, SingleAgentVoiceWorkflow, VoicePipeline, ) from agents.extensions.handoff_prompt import prompt_with_handoff_instructions @function_tool def get_weather(city: str) -> str: """Get the weather for a given city.""" print(f"[debug] get_weather called with city: {city}") choices = ["sunny", "cloudy", "rainy", "snowy"] return f"The weather in {city} is {random.choice(choices)}." spanish_agent = Agent( name="Spanish", handoff_description="A spanish speaking agent.", instructions=prompt_with_handoff_instructions( "You're speaking to a human, so be polite and concise. Speak in Spanish.", ), model="gpt-5.5", ) agent = Agent( name="Assistant", instructions=prompt_with_handoff_instructions( "You're speaking to a human, so be polite and concise. If the user speaks in Spanish, handoff to the spanish agent.", ), model="gpt-5.5", handoffs=[spanish_agent], tools=[get_weather], ) async def main(): pipeline = VoicePipeline(workflow=SingleAgentVoiceWorkflow(agent)) buffer = np.zeros(24000 * 3, dtype=np.int16) audio_input = AudioInput(buffer=buffer) result = await pipeline.run(audio_input) # Create an audio player using `sounddevice` player = sd.OutputStream(samplerate=24000, channels=1, dtype=np.int16) player.start() # Play the audio stream as it comes in async for event in result.stream(): if event.type == "voice_stream_event_audio": player.write(event.data) if __name__ == "__main__": asyncio.run(main()) ``` -------------------------------- ### Installation for Docker-backed sandboxes Source: https://openai.github.io/openai-agents-python/sandbox_agents Install the openai-agents package with Docker support. ```bash pip install "openai-agents[docker]" ``` -------------------------------- ### Installation Source: https://openai.github.io/openai-agents-python/realtime/quickstart Install the OpenAI Agents SDK using pip. ```bash pip install openai-agents ``` -------------------------------- ### Create a local sandbox agent Source: https://openai.github.io/openai-agents-python/sandbox_agents Example code to set up and run a local sandbox agent. ```python import asyncio from pathlib import Path from agents import Runner from agents.run import RunConfig from agents.sandbox import Manifest, SandboxAgent, SandboxRunConfig from agents.sandbox.capabilities import Capabilities, LocalDirLazySkillSource, Skills from agents.sandbox.entries import LocalDir from agents.sandbox.sandboxes.unix_local import UnixLocalSandboxClient EXAMPLE_DIR = Path(__file__).resolve().parent HOST_REPO_DIR = EXAMPLE_DIR / "repo" HOST_SKILLS_DIR = EXAMPLE_DIR / "skills" def build_agent(model: str) -> SandboxAgent[None]: return SandboxAgent( name="Sandbox engineer", model=model, instructions=( "Read `repo/task.md` before editing files. Stay grounded in the repository, preserve " "existing behavior, and mention the exact verification command you ran. " "If you edit files with apply_patch, paths are relative to the sandbox workspace root." ), default_manifest=Manifest( entries={ "repo": LocalDir(src=HOST_REPO_DIR), } ), capabilities=Capabilities.default() + [ Skills( lazy_from=LocalDirLazySkillSource( # This is a host path read by the SDK process. # Requested skills are copied into `skills_path` in the sandbox. source=LocalDir(src=HOST_SKILLS_DIR), ) ), ], ) async def main() -> None: result = await Runner.run( build_agent("gpt-5.5"), "Open `repo/task.md`, fix the issue, run the targeted test, and summarize the change.", run_config=RunConfig( sandbox=SandboxRunConfig(client=UnixLocalSandboxClient()), workflow_name="Sandbox coding example", ), ) print(result.final_output) if __name__ == "__main__": asyncio.run(main()) ``` -------------------------------- ### Start Session and Send Input Source: https://openai.github.io/openai-agents-python/realtime/quickstart Start the realtime session, send a message, and process events. ```python async def main() -> None: session = await runner.run() async with session: await session.send_message("Say hello in one short sentence.") async for event in session: if event.type == "audio": # Forward or play event.audio.data. pass elif event.type == "history_added": print(event.item) elif event.type == "agent_end": # One assistant turn finished. break elif event.type == "error": print(f"Error: {event.error}") if __name__ == "__main__": asyncio.run(main()) ``` -------------------------------- ### Install optional voice dependencies Source: https://openai.github.io/openai-agents-python/voice/quickstart Install the optional voice dependencies for the Agents SDK. ```bash pip install 'openai-agents[voice]' ``` -------------------------------- ### Sandbox Examples Source: https://openai.github.io/openai-agents-python/release Examples and tutorials for sandbox agents, covering various tasks and provider setups. ```python examples/sandbox/ ``` -------------------------------- ### Install OpenAI Agents SDK Source: https://openai.github.io/openai-agents-python/quickstart Command to install the OpenAI Agents SDK using pip or uv. ```bash pip install openai-agents # or `uv add openai-agents`, etc ``` -------------------------------- ### Define the Starting Agent Source: https://openai.github.io/openai-agents-python/realtime/quickstart Define the initial agent with its name and instructions. ```python agent = RealtimeAgent( name="Assistant", instructions="You are a helpful voice assistant. Keep responses short and conversational.", ) ``` -------------------------------- ### Hello world example Source: https://openai.github.io/openai-agents-python A basic example demonstrating how to create an agent and run a task synchronously. ```python from agents import Agent, Runner agent = Agent(name="Assistant", instructions="You are a helpful assistant") result = Runner.run_sync(agent, "Write a haiku about recursion in programming.") print(result.final_output) # Code within the code, # Functions calling themselves, # Infinite loop's dance. ``` -------------------------------- ### Quick Start with Database URL Source: https://openai.github.io/openai-agents-python/sessions/sqlalchemy_session Initialize a `SQLAlchemySession` using a database URL. This example uses an in-memory SQLite database. Ensure `create_tables=True` if tables do not exist. ```python import asyncio from agents import Agent, Runner from agents.extensions.memory import SQLAlchemySession async def main(): agent = Agent("Assistant") # Create session using database URL session = SQLAlchemySession.from_url( "user-123", url="sqlite+aiosqlite:///:memory:", create_tables=True ) result = await Runner.run(agent, "Hello", session=session) print(result.final_output) if __name__ == "__main__": asyncio.run(main()) ``` -------------------------------- ### Pass API Key Directly Source: https://openai.github.io/openai-agents-python/realtime/quickstart Pass the API key directly when starting the session. ```python session = await runner.run(model_config={"api_key": "your-api-key"}) ``` -------------------------------- ### Hosted Tool Search Example Source: https://openai.github.io/openai-agents-python/tools Example demonstrating how to use hosted tool search with deferred loading of tools and namespaces. ```python from typing import Annotated from agents import Agent, Runner, ToolSearchTool, function_tool, tool_namespace @function_tool(defer_loading=True) def get_customer_profile( customer_id: Annotated[str, "The customer ID to look up."], ) -> str: """Fetch a CRM customer profile.""" return f"profile for {customer_id}" @function_tool(defer_loading=True) def list_open_orders( customer_id: Annotated[str, "The customer ID to look up."], ) -> str: """List open orders for a customer.""" return f"open orders for {customer_id}" crm_tools = tool_namespace( name="crm", description="CRM tools for customer lookups.", tools=[get_customer_profile, list_open_orders], ) agent = Agent( name="Operations assistant", model="gpt-5.5", instructions="Load the crm namespace before using CRM tools.", tools=[*crm_tools, ToolSearchTool()], ) result = await Runner.run(agent, "Look up customer_42 and list their open orders.") print(result.final_output) ``` -------------------------------- ### Install openai-agents with encryption extra Source: https://openai.github.io/openai-agents-python/sessions/encrypted_session Install the library with the 'encrypt' extra to enable encrypted sessions. ```bash pip install openai-agents[encrypt] ``` -------------------------------- ### Install SQLAlchemy Extra Source: https://openai.github.io/openai-agents-python/sessions/sqlalchemy_session Install the `openai-agents` package with the `sqlalchemy` extra to enable SQLAlchemy session support. ```bash pip install openai-agents[sqlalchemy] ``` -------------------------------- ### Websocket Streaming Example Source: https://openai.github.io/openai-agents-python/release Example demonstrating websocket streaming, tools, approvals, and follow-up turns. ```python examples/basic/stream_ws.py ``` -------------------------------- ### Run the pipeline Source: https://openai.github.io/openai-agents-python/voice/quickstart Prepare audio input, run the pipeline, and play the resulting audio stream. ```python import numpy as np import sounddevice as sd from agents.voice import AudioInput # For simplicity, we'll just create 3 seconds of silence # In reality, you'd get microphone data buffer = np.zeros(24000 * 3, dtype=np.int16) audio_input = AudioInput(buffer=buffer) result = await pipeline.run(audio_input) # Create an audio player using `sounddevice` player = sd.OutputStream(samplerate=24000, channels=1, dtype=np.int16) player.start() # Play the audio stream as it comes in async for event in result.stream(): if event.type == "voice_stream_event_audio": player.write(event.data) ``` -------------------------------- ### Prefetching all MCP Tools Example Source: https://openai.github.io/openai-agents-python/examples Shows how to prefetch all available MCP tools using `MCPUtil.get_all_function_tools`. This can be useful for agents that need a comprehensive list of available tools upfront. ```python from openai_agents.mcp import MCPUtil async def main(): # Replace with the URL of your MCP server mcp_server_url = "http://localhost:8000/mcp" try: tools = await MCPUtil.get_all_function_tools(mcp_server_url) print(f"Fetched {len(tools)} tools:") for tool in tools: print(f"- {tool['name']}") except Exception as e: print(f"Could not connect to MCP server at {mcp_server_url}. Ensure it is running. Error: {e}") if __name__ == "__main__": import asyncio asyncio.run(main()) ``` -------------------------------- ### SIP Integration Example Source: https://openai.github.io/openai-agents-python/realtime/guide Example of using OpenAIRealtimeSIPModel to attach an agent session to a call arriving via the Realtime Calls API. ```python from agents.realtime import RealtimeRunner from agents.realtime.openai_realtime import OpenAIRealtimeSIPModel runner = RealtimeRunner(starting_agent=agent, model=OpenAIRealtimeSIPModel()) async with await runner.run( model_config={ "call_id": call_id_from_webhook, } ) as session: async for event in session: ... ``` -------------------------------- ### Quick start with AdvancedSQLiteSession Source: https://openai.github.io/openai-agents-python/sessions/advanced_sqlite_session Demonstrates initializing an agent with an advanced session and performing conversation turns while tracking usage. ```python from agents import Agent, Runner from agents.extensions.memory import AdvancedSQLiteSession # Create agent agent = Agent( name="Assistant", instructions="Reply very concisely.", ) # Create an advanced session session = AdvancedSQLiteSession( session_id="conversation_123", db_path="conversations.db", create_tables=True ) # First conversation turn result = await Runner.run( agent, "What city is the Golden Gate Bridge in?", session=session ) print(result.final_output) # "San Francisco" # IMPORTANT: Store usage data await session.store_run_usage(result) # Continue conversation result = await Runner.run( agent, "What state is it in?", session=session ) print(result.final_output) # "California" await session.store_run_usage(result) ``` -------------------------------- ### Sensitive Data Guardrail Example Source: https://openai.github.io/openai-agents-python/realtime/guide Example of defining a guardrail function to check for sensitive data (like passwords) in agent output. ```python from agents.guardrail import GuardrailFunctionOutput, OutputGuardrail def sensitive_data_check(context, agent, output): return GuardrailFunctionOutput( tripwire_triggered="password" in output, output_info=None, ) agent = RealtimeAgent( name="Assistant", instructions="...", output_guardrails=[OutputGuardrail(guardrail_function=sensitive_data_check)], ) ``` -------------------------------- ### Create Project and Virtual Environment Source: https://openai.github.io/openai-agents-python/quickstart Commands to create a new project directory and set up a Python virtual environment. ```bash mkdir my_project cd my_project python -m venv .venv ``` -------------------------------- ### HTTP with SSE MCP Server Example Source: https://openai.github.io/openai-agents-python/mcp An example demonstrating the setup and usage of an HTTP with SSE MCP server. ```python from agents import Agent, Runner from agents.model_settings import ModelSettings from agents.mcp import MCPServerSse workspace_id = "demo-workspace" async with MCPServerSse( name="SSE Python Server", params={ "url": "http://localhost:8000/sse", "headers": {"X-Workspace": workspace_id}, }, cache_tools_list=True, ) as server: agent = Agent( name="Assistant", mcp_servers=[server], model_settings=ModelSettings(tool_choice="required"), ) result = await Runner.run(agent, "What's the weather in Tokyo?") print(result.final_output) ``` -------------------------------- ### Defining a Function Tool Source: https://openai.github.io/openai-agents-python/realtime/guide Example of defining a function tool using the `@function_tool` decorator. ```python from agents import function_tool @function_tool def get_weather(city: str) -> str: """Get current weather for a city.""" return f"The weather in {city} is sunny, 72F." agent = RealtimeAgent( name="Assistant", instructions="You can answer weather questions.", tools=[get_weather], ) ``` -------------------------------- ### Import Realtime Components Source: https://openai.github.io/openai-agents-python/realtime/quickstart Import necessary components for creating a server-side realtime session. ```python import asyncio from agents.realtime import RealtimeAgent, RealtimeRunner ``` -------------------------------- ### Example Usage of WebSearchTool and FileSearchTool Source: https://openai.github.io/openai-agents-python/tools This snippet demonstrates how to initialize an Agent with WebSearchTool and FileSearchTool, and then run a query using the Runner. ```python 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) ``` -------------------------------- ### Defining Agents with Handoffs Source: https://openai.github.io/openai-agents-python/realtime/guide Example of defining two `RealtimeAgent` instances, one with a handoff to another specialist agent. ```python from agents.realtime import RealtimeAgent, realtime_handoff billing_agent = RealtimeAgent( name="Billing Support", instructions="You specialize in billing issues.", ) main_agent = RealtimeAgent( name="Customer Service", instructions="Triage the request and hand off when needed.", handoffs=[realtime_handoff(billing_agent, tool_description="Transfer to billing support")], ) ``` -------------------------------- ### Enable Memory Capability for Sandbox Agent Source: https://openai.github.io/openai-agents-python/sandbox/memory Add `Memory()` as a capability when initializing a `SandboxAgent`. This example shows the basic setup with `Filesystem()` and `Shell()` capabilities. ```python from pathlib import Path import tempfile from agents.sandbox import LocalSnapshotSpec, SandboxAgent from agents.sandbox.capabilities import Filesystem, Memory, Shell agent = SandboxAgent( name="Memory-enabled reviewer", instructions="Inspect the workspace and preserve useful lessons for follow-up runs.", capabilities=[Memory(), Filesystem(), Shell()], ) with tempfile.TemporaryDirectory(prefix="sandbox-memory-example-") as snapshot_dir: sandbox = await client.create( manifest=manifest, snapshot=LocalSnapshotSpec(base_path=Path(snapshot_dir)), ) ``` -------------------------------- ### MCPServerManager with FastAPI Example Source: https://openai.github.io/openai-agents-python/examples Demonstrates how to set up an MCP server using `MCPServerManager` with FastAPI. This provides a framework for building MCP-compatible services. ```python from fastapi import FastAPI from openai_agents.mcp import MCPServerManager, tool class MyTools: @tool def greet(self, name: str) -> str: """Greets the user.""" return f"Hello, {name}!" app = FastAPI() m = MCPServerManager(MyTools()) @app.websocket("/mcp") async def websocket_endpoint(websocket: fastapi.WebSocket): await m.handle_websocket(websocket) # To run this: # 1. Save as main.py # 2. Run: uvicorn main:app --reload # 3. Connect using an MCPClient ``` -------------------------------- ### Azure OpenAI Connection with API Key Source: https://openai.github.io/openai-agents-python/realtime/guide Example of connecting to Azure OpenAI using a specific endpoint URL and API key. ```python session = await runner.run( model_config={ "url": "wss://.openai.azure.com/openai/v1/realtime?model=", "headers": {"api-key": ""}, } ) ``` -------------------------------- ### Implement the on_start Async Method Source: https://openai.github.io/openai-agents-python/ref/voice/workflow Use on_start to perform initialization tasks or deliver initial messages before receiving user input. ```python async def on_start(self) -> AsyncIterator[str]: """ Optional method that runs before any user input is received. Can be used to deliver a greeting or instruction via TTS. Defaults to doing nothing. """ return yield ``` -------------------------------- ### Run Your First Agent Source: https://openai.github.io/openai-agents-python/quickstart Python code to run the agent and print the result of a query. ```python import asyncio from agents import Agent, Runner agent = Agent( name="History Tutor", instructions="You answer history questions clearly and concisely.", ) async def main(): result = await Runner.run(agent, "When did the Roman Empire fall?") print(result.final_output) if __name__ == "__main__": asyncio.run(main()) ``` -------------------------------- ### Azure OpenAI Connection with Bearer Token Source: https://openai.github.io/openai-agents-python/realtime/guide Example of connecting to Azure OpenAI using a specific endpoint URL and token-based authentication (Bearer token). ```python session = await runner.run( model_config={ "url": "wss://.openai.azure.com/openai/v1/realtime?model=", "headers": {"authorization": f"Bearer {token}"}, } ) ``` -------------------------------- ### Provider-level setup for WebSocket transport Source: https://openai.github.io/openai-agents-python/models This example demonstrates configuring WebSocket transport at the provider level, including optional base URL and keepalive settings. ```python from agents import Agent, OpenAIProvider, RunConfig, Runner provider = OpenAIProvider( use_responses_websocket=True, # Optional; if omitted, OPENAI_WEBSOCKET_BASE_URL is used when set. websocket_base_url="wss://your-proxy.example/v1", # Optional low-level websocket keepalive settings. responses_websocket_options={"ping_interval": 20.0, "ping_timeout": 60.0}, ) agent = Agent(name="Assistant") result = await Runner.run( agent, "Hello", run_config=RunConfig(model_provider=provider), ) ``` -------------------------------- ### Initialize and Run a RealtimeSession Source: https://openai.github.io/openai-agents-python/ref/realtime/runner Demonstrates how to instantiate a RealtimeRunner and manage a session lifecycle using an asynchronous context manager. ```python runner = RealtimeRunner(agent) async with await runner.run() as session: await session.send_message("Hello") async for event in session: print(event) ``` -------------------------------- ### Developer-owned lifecycle example using direct lifecycle methods Source: https://openai.github.io/openai-agents-python/sandbox/guide Illustrates managing a developer-owned sandbox lifecycle by directly calling lifecycle methods like `create`, `start`, `stop`, and `aclose`. ```python sandbox = await client.create( manifest=agent.default_manifest, snapshot=LocalSnapshotSpec(base_path=Path("/tmp/my-sandbox-snapshots")), ) try: await sandbox.start() await Runner.run( agent, "Analyze the files.", run_config=RunConfig(sandbox=SandboxRunConfig(session=sandbox)), ) # Persist a checkpoint of the live workspace before doing more work. # `aclose()` also calls `stop()`, so this is only needed for an explicit mid-lifecycle save. await sandbox.stop() finally: await sandbox.aclose() ``` -------------------------------- ### Give Your Agent Tools Source: https://openai.github.io/openai-agents-python/quickstart Python code demonstrating how to add a tool to an agent and use it. ```python import asyncio from agents import Agent, Runner, function_tool @function_tool def history_fun_fact() -> str: """Return a short history fact.""" return "Sharks are older than trees." agent = Agent( name="History Tutor", instructions="Answer history questions clearly. Use history_fun_fact when it helps.", tools=[history_fun_fact], ) async def main(): result = await Runner.run( agent, "Tell me something surprising about ancient life on Earth.", ) print(result.final_output) if __name__ == "__main__": asyncio.run(main()) ``` -------------------------------- ### get_trace_provider function Source: https://openai.github.io/openai-agents-python/ref/tracing/setup Gets the global trace provider used by tracing utilities. The default provider and processor are initialized lazily on first access so importing the SDK does not create network clients or threading primitives. ```python def get_trace_provider() -> TraceProvider: """Get the global trace provider used by tracing utilities. The default provider and processor are initialized lazily on first access so importing the SDK does not create network clients or threading primitives. """ global GLOBAL_TRACE_PROVIDER global _SHUTDOWN_HANDLER_REGISTERED provider = GLOBAL_TRACE_PROVIDER if provider is not None: return provider with _GLOBAL_TRACE_PROVIDER_LOCK: provider = GLOBAL_TRACE_PROVIDER if provider is None: from .processors import default_processor from .provider import DefaultTraceProvider provider = DefaultTraceProvider() provider.register_processor(default_processor()) GLOBAL_TRACE_PROVIDER = provider if not _SHUTDOWN_HANDLER_REGISTERED: atexit.register(_shutdown_global_trace_provider) _SHUTDOWN_HANDLER_REGISTERED = True return provider ``` -------------------------------- ### Simple Hosted MCP Example Source: https://openai.github.io/openai-agents-python/examples Demonstrates the basic usage of hosted MCP (Model Context Protocol) without requiring explicit approval for tool calls. Suitable for automated workflows. ```python from openai_agents import Agent, tool class SimpleMCPAgent(Agent): @tool def get_time(self) -> str: """Returns the current time.""" import datetime return datetime.datetime.now().isoformat() def main(): agent = SimpleMCPAgent() response = agent.run("What time is it?") print(f"Response: {response}") if __name__ == "__main__": main() ``` -------------------------------- ### CustomProcessor Example Source: https://openai.github.io/openai-agents-python/ref/tracing An example implementation of a custom TracingProcessor. ```python class CustomProcessor(TracingProcessor): def __init__(self): self.active_traces = {} self.active_spans = {} def on_trace_start(self, trace): self.active_traces[trace.trace_id] = trace def on_trace_end(self, trace): # Process completed trace del self.active_traces[trace.trace_id] def on_span_start(self, span): self.active_spans[span.span_id] = span def on_span_end(self, span): # Process completed span del self.active_spans[span.span_id] def shutdown(self): # Clean up resources self.active_traces.clear() self.active_spans.clear() def force_flush(self): # Force processing of any queued items pass ``` -------------------------------- ### Codex tool usage example Source: https://openai.github.io/openai-agents-python/tools Example of how to initialize and use the `codex_tool` with an Agent. ```python from agents import Agent from agents.extensions.experimental.codex import ThreadOptions, TurnOptions, codex_tool agent = Agent( name="Codex Agent", instructions="Use the codex tool to inspect the workspace and answer the question.", tools=[ codex_tool( sandbox_mode="workspace-write", working_directory="/path/to/repo", default_thread_options=ThreadOptions( model="gpt-5.5", model_reasoning_effort="low", network_access_enabled=True, web_search_mode="disabled", approval_policy="never", ), default_turn_options=TurnOptions( idle_timeout_seconds=60, ), persist_session=True, ) ], ) ``` -------------------------------- ### Create Your First Agent Source: https://openai.github.io/openai-agents-python/quickstart Python code to define a simple agent with a name and instructions. ```python from agents import Agent agent = Agent( name="History Tutor", instructions="You answer history questions clearly and concisely.", ) ``` -------------------------------- ### MongoDB Sessions Installation Source: https://openai.github.io/openai-agents-python/sessions Install the necessary package for MongoDB session integration. ```bash pip install openai-agents[mongodb] ``` -------------------------------- ### Set up Voice Pipeline Source: https://openai.github.io/openai-agents-python/voice/quickstart Initialize a VoicePipeline using SingleAgentVoiceWorkflow with the defined agent. ```python from agents.voice import SingleAgentVoiceWorkflow, VoicePipeline pipeline = VoicePipeline(workflow=SingleAgentVoiceWorkflow(agent)) ``` -------------------------------- ### Dapr Sessions Installation Source: https://openai.github.io/openai-agents-python/sessions Install the necessary package for Dapr session integration. ```bash pip install openai-agents[dapr] ``` -------------------------------- ### Set up Agents Source: https://openai.github.io/openai-agents-python/voice/quickstart Define agents, including a Spanish-speaking agent and a main assistant agent with tools and handoff capabilities. ```python import asyncio import random from agents import ( Agent, function_tool, ) from agents.extensions.handoff_prompt import prompt_with_handoff_instructions @function_tool def get_weather(city: str) -> str: """Get the weather for a given city.""" print(f"[debug] get_weather called with city: {city}") choices = ["sunny", "cloudy", "rainy", "snowy"] return f"The weather in {city} is {random.choice(choices)}." spanish_agent = Agent( name="Spanish", handoff_description="A spanish speaking agent.", instructions=prompt_with_handoff_instructions( "You're speaking to a human, so be polite and concise. Speak in Spanish.", ), model="gpt-5.5", ) agent = Agent( name="Assistant", instructions=prompt_with_handoff_instructions( "You're speaking to a human, so be polite and concise. If the user speaks in Spanish, handoff to the spanish agent.", ), model="gpt-5.5", handoffs=[spanish_agent], tools=[get_weather], ) ``` -------------------------------- ### MCP Connectors Example (Google Calendar) Source: https://openai.github.io/openai-agents-python/examples Shows how to integrate MCP with external connectors, using Google Calendar as an example. Requires setting up Google Calendar API credentials. ```python from openai_agents import Agent, tool from openai_agents.mcp.connectors.google_calendar import GoogleCalendarConnector class CalendarAgent(Agent): def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) self.add_connector(GoogleCalendarConnector()) @tool def create_calendar_event(self, summary: str, start_time: str, end_time: str) -> str: """Creates a Google Calendar event.""" # This tool is a placeholder; the actual creation is handled by the connector return f"Event '{summary}' created from {start_time} to {end_time}." def main(): agent = CalendarAgent() # Example usage (requires Google Calendar setup) # response = agent.run("Create a meeting about project sync tomorrow from 2pm to 3pm.") # print(response) print("Google Calendar connector added. Uncomment and configure to use.") if __name__ == "__main__": main() ``` -------------------------------- ### Redis sessions installation Source: https://openai.github.io/openai-agents-python/sessions Command to install the Redis extra for the Agents SDK. ```bash pip install openai-agents[redis] ``` -------------------------------- ### start method Source: https://openai.github.io/openai-agents-python/ref/tracing Start the trace and optionally mark it as the current trace. ```python @abc.abstractmethod def start(self, mark_as_current: bool = False): """Start the trace and optionally mark it as the current trace. Args: mark_as_current: If true, marks this trace as the current trace in the execution context. Notes: - Must be called before any spans can be added - Only one trace can be current at a time - Thread-safe when using mark_as_current """ pass ``` -------------------------------- ### Define start method Source: https://openai.github.io/openai-agents-python/ref/tracing/spans Starts the span execution, optionally marking it as the current span. ```python start(mark_as_current: bool = False) ``` ```python @abc.abstractmethod def start(self, mark_as_current: bool = False): """ Start the span. Args: mark_as_current: If true, the span will be marked as the current span. """ pass ``` -------------------------------- ### Quick Start with Existing Engine Source: https://openai.github.io/openai-agents-python/sessions/sqlalchemy_session Integrate `SQLAlchemySession` with an existing SQLAlchemy asynchronous engine. Remember to dispose of the engine when done. ```python import asyncio from agents import Agent, Runner from agents.extensions.memory import SQLAlchemySession from sqlalchemy.ext.asyncio import create_async_engine async def main(): # Create your database engine engine = create_async_engine("postgresql+asyncpg://user:pass@localhost/db") agent = Agent("Assistant") session = SQLAlchemySession( "user-456", engine=engine, create_tables=True ) result = await Runner.run(agent, "Hello", session=session) print(result.final_output) # Clean up await engine.dispose() if __name__ == "__main__": asyncio.run(main()) ``` -------------------------------- ### Configure the Runner Source: https://openai.github.io/openai-agents-python/realtime/quickstart Configure the RealtimeRunner with agent and model settings, including audio input/output formats and transcription models. ```python runner = RealtimeRunner( starting_agent=agent, config={ "model_settings": { "model_name": "gpt-realtime-2", "audio": { "input": { "format": "pcm16", "transcription": {"model": "gpt-4o-mini-transcribe"}, "turn_detection": { "type": "semantic_vad", "interrupt_response": True, }, }, "output": { "format": "pcm16", "voice": "ash", }, }, } }, ) ``` -------------------------------- ### Full example: coding task Source: https://openai.github.io/openai-agents-python/sandbox/guide A comprehensive example demonstrating how to set up and run a SandboxAgent for a coding task. This example includes defining the agent, its capabilities, and the execution flow. ```python import asyncio from pathlib import Path from agents import ModelSettings, Runner from agents.run import RunConfig from agents.sandbox import Manifest, SandboxAgent, SandboxRunConfig from agents.sandbox.capabilities import ( Capabilities, LocalDirLazySkillSource, Skills, ) from agents.sandbox.entries import LocalDir from agents.sandbox.sandboxes.unix_local import UnixLocalSandboxClient EXAMPLE_DIR = Path(__file__).resolve().parent HOST_REPO_DIR = EXAMPLE_DIR / "repo" HOST_SKILLS_DIR = EXAMPLE_DIR / "skills" TARGET_TEST_CMD = "sh tests/test_credit_note.sh" def build_agent(model: str) -> SandboxAgent[None]: return SandboxAgent( name="Sandbox engineer", model=model, instructions=( "Inspect the repo, make the smallest correct change, run the most relevant checks, " "and summarize the file changes and risks. " "Read `repo/task.md` before editing files. Stay grounded in the repository, preserve " "existing behavior, and mention the exact verification command you ran. " "Use the `$credit-note-fixer` skill before editing files. If the repo lives under " "`repo/`, remember that `apply_patch` paths stay relative to the sandbox workspace " "root, so edits still target `repo/...`." ), # Put repos and task files in the manifest. default_manifest=Manifest( entries={ "repo": LocalDir(src=HOST_REPO_DIR), } ), capabilities=Capabilities.default() + [ Skills( lazy_from=LocalDirLazySkillSource( # This is a host path read by the SDK process. # Requested skills are copied into `skills_path` in the sandbox. source=LocalDir(src=HOST_SKILLS_DIR), ) ), ], model_settings=ModelSettings(tool_choice="required"), ) async def main(model: str, prompt: str) -> None: result = await Runner.run( build_agent(model), prompt, run_config=RunConfig( sandbox=SandboxRunConfig(client=UnixLocalSandboxClient()), workflow_name="Sandbox coding example", ), ) print(result.final_output) if __name__ == "__main__": asyncio.run( main( model="gpt-5.5", prompt=( "Open `repo/task.md`, use the `$credit-note-fixer` skill, fix the bug, " f"run `{TARGET_TEST_CMD}`, and summarize the change." ), ) ) ``` -------------------------------- ### Quick start Source: https://openai.github.io/openai-agents-python/sessions Demonstrates how to use sessions for maintaining conversation history across multiple agent runs, including both asynchronous and synchronous runners. ```python from agents import Agent, Runner, SQLiteSession # Create agent agent = Agent( name="Assistant", instructions="Reply very concisely.", ) # Create a session instance with a session ID session = SQLiteSession("conversation_123") # First turn result = await Runner.run( agent, "What city is the Golden Gate Bridge in?", session=session ) print(result.final_output) # "San Francisco" # Second turn - agent automatically remembers previous context result = await Runner.run( agent, "What state is it in?", session=session ) print(result.final_output) # "California" # Also works with synchronous runner result = Runner.run_sync( agent, "What's the population?", session=session ) print(result.final_output) # "Approximately 39 million" ``` -------------------------------- ### RealtimeSession Example Source: https://openai.github.io/openai-agents-python/ref/realtime/session Example of how to use RealtimeSession to send messages and audio, and stream events. ```python runner = RealtimeRunner(agent) async with await runner.run() as session: # Send messages await session.send_message("Hello") await session.send_audio(audio_bytes) # Stream events async for event in session: if event.type == "audio": # Handle audio event pass ``` -------------------------------- ### Async SQLite sessions installation Source: https://openai.github.io/openai-agents-python/sessions Command to install aiosqlite for Async SQLite sessions. ```bash pip install aiosqlite ``` -------------------------------- ### Activate Virtual Environment (macOS/Linux) Source: https://openai.github.io/openai-agents-python/quickstart Command to activate the virtual environment on macOS or Linux systems. ```bash source .venv/bin/activate ``` -------------------------------- ### Example Usage Source: https://openai.github.io/openai-agents-python/ref/realtime/session Demonstrates how to use RealtimeSession to send messages and audio, and how to stream events from the model. ```python runner = RealtimeRunner(agent) async with await runner.run() as session: # Send messages await session.send_message("Hello") await session.send_audio(audio_bytes) # Stream events async for event in session: if event.type == "audio": # Handle audio event pass ``` -------------------------------- ### Custom Span Example Source: https://openai.github.io/openai-agents-python/ref/tracing Example of creating and using a custom span for a database query. ```python # Creating a custom span with custom_span("database_query", { "operation": "SELECT", "table": "users" }) as span: results = await db.query("SELECT * FROM users") span.set_output({"count": len(results)}) ```