OpenAI Agents SDK

from agents import Agent, Runner

# Tools
from agents import function_tool, FunctionTool, FileSearchTool, WebSearchTool, ComputerTool

# Handoffs
from agents import Handoff, handoff

# Guardrails
from agents import InputGuardrail, OutputGuardrail, input_guardrail, output_guardrail

# Memory/Sessions
from agents import Session, SQLiteSession, OpenAIConversationsSession

# Model Configuration
from agents import ModelSettings, RunConfig

# Results and Items
from agents import RunResult, RunResultStreaming, RunItem, ModelResponse

# Tracing
from agents.tracing import trace, Trace, Span

# MCP
from agents.mcp import MCPServer, MCPServerStdio

# Realtime
from agents.realtime import RealtimeAgent, RealtimeRunner

# Voice
from agents.voice import VoicePipeline, STTModel, TTSModel

from agents import Agent, Runner

# Create a simple agent
agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant"
)

# Run synchronously
result = Runner.run_sync(agent, "Write a haiku about recursion in programming.")
print(result.final_output)

# Run asynchronously
import asyncio

async def main():
    result = await Runner.run(agent, "What is the weather like today?")
    print(result.final_output)

asyncio.run(main())

from agents import Agent, Runner, function_tool

@function_tool
def get_weather(city: str) -> str:
    """Get the weather for a city."""
    return f"The weather in {city} is sunny."

agent = Agent(
    name="Weather Agent",
    instructions="You help users check the weather.",
    tools=[get_weather]
)

result = Runner.run_sync(agent, "What's the weather in Tokyo?")
print(result.final_output)  # The weather in Tokyo is sunny.

from agents import Agent, Runner

spanish_agent = Agent(
    name="Spanish Agent",
    instructions="You only speak Spanish."
)

english_agent = Agent(
    name="English Agent",
    instructions="You only speak English"
)

triage_agent = Agent(
    name="Triage Agent",
    instructions="Handoff to the appropriate agent based on language.",
    handoffs=[spanish_agent, english_agent]
)

result = Runner.run_sync(triage_agent, "Hola, ¿cómo estás?")
print(result.final_output)  # ¡Hola! Estoy bien, gracias...

class Agent[TContext]:
    name: str
    instructions: str | Callable | None
    prompt: Prompt | DynamicPromptFunction | None
    tools: list[Tool]
    handoffs: list[Agent | Handoff]
    model: str | Model | None
    model_settings: ModelSettings
    mcp_servers: list[MCPServer]
    mcp_config: MCPConfig
    input_guardrails: list[InputGuardrail]
    output_guardrails: list[OutputGuardrail]
    output_type: type[Any] | AgentOutputSchemaBase | None
    hooks: AgentHooks | None
    tool_use_behavior: Literal | StopAtTools | ToolsToFinalOutputFunction
    reset_tool_choice: bool
    handoff_description: str | None

    def clone(**kwargs) -> Agent: ...
    def as_tool(...) -> Tool: ...
    def get_system_prompt(context) -> str | None: ...
    def get_all_tools(context) -> list[Tool]: ...

class Runner:
    @classmethod
    async def run(starting_agent, input, *, context, max_turns, hooks,
                  run_config, previous_response_id, conversation_id,
                  session) -> RunResult: ...

    @classmethod
    def run_sync(...) -> RunResult: ...

    @classmethod
    def run_streamed(...) -> RunResultStreaming: ...

class RunConfig:
    model: str | Model | None
    model_provider: ModelProvider
    model_settings: ModelSettings | None
    handoff_input_filter: HandoffInputFilter | None
    nest_handoff_history: bool
    input_guardrails: list[InputGuardrail] | None
    output_guardrails: list[OutputGuardrail] | None
    tracing_disabled: bool
    workflow_name: str
    trace_id: str | None
    ...

@function_tool
def my_function(param: str) -> str:
    """Function description."""
    ...

class FunctionTool:
    name: str
    description: str
    params_json_schema: dict[str, Any]
    on_invoke_tool: Callable
    strict_json_schema: bool
    is_enabled: bool | Callable
    tool_input_guardrails: list[ToolInputGuardrail] | None
    tool_output_guardrails: list[ToolOutputGuardrail] | None

class FileSearchTool:
    vector_store_ids: list[str]
    max_num_results: int | None
    include_search_results: bool
    ranking_options: RankingOptions | None
    filters: Filters | None

class WebSearchTool:
    user_location: UserLocation | None
    filters: WebSearchToolFilters | None
    search_context_size: Literal["low", "medium", "high"]

class ComputerTool:
    computer: Computer | AsyncComputer
    on_safety_check: Callable | None

class ShellTool:
    executor: ShellExecutor
    name: str

class ApplyPatchTool:
    editor: ApplyPatchEditor
    name: str

class HostedMCPTool:
    tool_config: Mcp
    on_approval_request: MCPToolApprovalFunction | None

class CodeInterpreterTool:
    tool_config: CodeInterpreter

class ImageGenerationTool:
    tool_config: ImageGeneration

class LocalShellTool:
    executor: LocalShellExecutor

class Handoff[TContext, TAgent]:
    tool_name: str
    tool_description: str
    input_json_schema: dict[str, Any]
    on_invoke_handoff: Callable
    agent_name: str
    input_filter: HandoffInputFilter | None
    nest_handoff_history: bool | None
    strict_json_schema: bool
    is_enabled: bool | Callable

    def get_transfer_message(agent) -> str: ...

def handoff(agent, tool_name_override, tool_description_override,
            on_handoff, input_type, input_filter, nest_handoff_history,
            is_enabled) -> Handoff: ...

class HandoffInputData:
    input_history: str | tuple[TResponseInputItem, ...]
    pre_handoff_items: tuple[RunItem, ...]
    new_items: tuple[RunItem, ...]
    run_context: RunContextWrapper | None

    def clone(**kwargs) -> HandoffInputData: ...

HandoffHistoryMapper = Callable[[list[TResponseInputItem]], list[TResponseInputItem]]

def nest_handoff_history() -> list[TResponseInputItem]: ...
def default_handoff_history_mapper() -> list[TResponseInputItem]: ...

class InputGuardrail[TContext]:
    guardrail_function: Callable
    name: str | None
    run_in_parallel: bool

    def get_name() -> str: ...
    async def run(agent, input, context) -> InputGuardrailResult: ...

class OutputGuardrail[TContext]:
    guardrail_function: Callable
    name: str | None

    def get_name() -> str: ...
    async def run(context, agent, agent_output) -> OutputGuardrailResult: ...

@input_guardrail
def my_input_check(input: str) -> GuardrailFunctionOutput:
    """Check input before agent processes it."""
    ...

@output_guardrail
def my_output_check(output: str) -> GuardrailFunctionOutput:
    """Check output before returning to user."""
    ...

class ToolInputGuardrail[TContext]:
    guardrail_function: Callable
    name: str | None

    async def run(data) -> ToolGuardrailFunctionOutput: ...

class ToolOutputGuardrail[TContext]:
    guardrail_function: Callable
    name: str | None

    async def run(data) -> ToolGuardrailFunctionOutput: ...

class GuardrailFunctionOutput:
    output_info: Any
    tripwire_triggered: bool

class ToolGuardrailFunctionOutput:
    output_info: Any
    behavior: RejectContentBehavior | RaiseExceptionBehavior | AllowBehavior

    @classmethod
    def allow(output_info) -> ToolGuardrailFunctionOutput: ...

    @classmethod
    def reject_content(message, output_info) -> ToolGuardrailFunctionOutput: ...

    @classmethod
    def raise_exception(output_info) -> ToolGuardrailFunctionOutput: ...

class SessionABC:
    async def get_items() -> list[TResponseInputItem]: ...
    async def add_items(items) -> None: ...
    async def clear() -> None: ...

class SQLiteSession(SessionABC):
    def __init__(session_id, db_path): ...

class OpenAIConversationsSession(SessionABC):
    def __init__(conversation_id, client): ...

# In agents.extensions.memory
class RedisSession(SessionABC): ...
class SQLAlchemySession(SessionABC): ...
class AdvancedSQLiteSession(SessionABC): ...
class DaprSession(SessionABC): ...
class EncryptSession: ...  # Wrapper for encrypted sessions

class Model:
    async def get_response(...) -> ModelResponse: ...
    async def stream_response(...) -> AsyncIterator[TResponseStreamEvent]: ...

class ModelProvider:
    def get_model(model_name) -> Model: ...

class OpenAIProvider(ModelProvider):
    def get_model(model_name) -> Model: ...

class MultiProvider(ModelProvider):
    def get_model(model_name) -> Model: ...

class OpenAIChatCompletionsModel(Model): ...
class OpenAIResponsesModel(Model): ...

class ModelSettings:
    temperature: float | None
    top_p: float | None
    frequency_penalty: float | None
    presence_penalty: float | None
    tool_choice: ToolChoice | None
    parallel_tool_calls: bool | None
    max_tokens: int | None
    reasoning: Reasoning | None
    verbosity: Literal["low", "medium", "high"] | None
    ...

    def resolve(override) -> ModelSettings: ...

# In agents.extensions.models
class LiteLLMProvider(ModelProvider): ...

class Trace:
    trace_id: str
    name: str
    group_id: str | None
    metadata: dict[str, Any] | None

    def start(mark_as_current) -> None: ...
    def finish(reset_current) -> None: ...

class Span[TSpanData]:
    span_id: str
    name: str
    span_data: TSpanData
    parent_span: Span | None

    def start(mark_as_current) -> None: ...
    def finish(reset_current) -> None: ...

def trace(workflow_name, *, trace_id, group_id, metadata, disabled) -> Trace: ...

def agent_span(name, *, handoffs, output_type) -> Span[AgentSpanData]: ...
def function_span(name, *, input, output) -> Span[FunctionSpanData]: ...
def generation_span(name, *, model, input, output, usage) -> Span[GenerationSpanData]: ...
def guardrail_span(name, *, guardrail_type, input, output) -> Span[GuardrailSpanData]: ...
def handoff_span(name, *, from_agent, to_agent) -> Span[HandoffSpanData]: ...
def response_span(name, *, response_data) -> Span[ResponseSpanData]: ...
def speech_span(name, *, speech_data) -> Span[SpeechSpanData]: ...
def speech_group_span(name, *, group_data) -> Span[SpeechGroupSpanData]: ...
def transcription_span(name, *, transcription_data) -> Span[TranscriptionSpanData]: ...
def mcp_tools_span(name, *, server_label, tools) -> Span[MCPListToolsSpanData]: ...
def custom_span(name, *, custom_data) -> Span[CustomSpanData]: ...

class SpanData(abc.ABC):
    """Base class for span data types."""
    def export() -> dict[str, Any]: ...
    @property
    def type() -> str: ...

class AgentSpanData(SpanData): ...
class CustomSpanData(SpanData): ...
class FunctionSpanData(SpanData): ...
class GenerationSpanData(SpanData): ...
class GuardrailSpanData(SpanData): ...
class HandoffSpanData(SpanData): ...
class MCPListToolsSpanData(SpanData): ...
class SpeechSpanData(SpanData): ...
class SpeechGroupSpanData(SpanData): ...
class TranscriptionSpanData(SpanData): ...

class SpanError:
    message: str
    data: dict | None

def get_current_trace() -> Trace | None: ...
def get_current_span() -> Span | None: ...

class TracingProcessor:
    async def on_trace_start(trace) -> None: ...
    async def on_trace_end(trace) -> None: ...
    async def on_span_start(span) -> None: ...
    async def on_span_end(span) -> None: ...

def add_trace_processor(processor: TracingProcessor) -> None: ...
def set_trace_processors(processors: list[TracingProcessor]) -> None: ...
def set_tracing_disabled(disabled: bool) -> None: ...
def set_tracing_export_api_key(api_key: str) -> None: ...

class TraceProvider:
    """Provider for trace creation."""
    def create_trace(...) -> Trace: ...
    def create_span(...) -> Span: ...
    def register_processor(processor) -> None: ...
    def set_processors(processors) -> None: ...
    def set_disabled(disabled) -> None: ...
    def shutdown() -> None: ...

class DefaultTraceProvider(TraceProvider):
    """Default trace provider implementation."""
    ...

def get_trace_provider() -> TraceProvider: ...

class RealtimeAgent:
    """Agent for real-time audio interactions."""
    ...

class RealtimeRunner:
    """Runner for realtime agents."""
    @classmethod
    async def run(...): ...

class RealtimeSession:
    """Session for realtime interactions."""
    ...

class VoicePipeline:
    """Pipeline for voice processing."""
    ...

class STTModel:
    """Speech-to-text model interface."""
    async def transcribe(...): ...

class TTSModel:
    """Text-to-speech model interface."""
    async def synthesize(...): ...

class MCPServerStdio:
    def __init__(params: MCPServerStdioParams): ...
    async def connect(): ...
    async def cleanup(): ...

class MCPServerSse:
    def __init__(params: MCPServerSseParams): ...

class MCPServerStreamableHttp:
    def __init__(params: MCPServerStreamableHttpParams): ...

class MCPUtil:
    @staticmethod
    async def get_all_function_tools(...) -> list[Tool]: ...

def create_static_tool_filter(allowed_tools, denied_tools) -> ToolFilterStatic: ...

class ToolFilterContext:
    tool_name: str
    server_label: str

class MessageOutputItem:
    raw_item: ResponseOutputMessage
    agent: Agent
    type: Literal["message_output_item"]

class ToolCallItem:
    raw_item: ToolCallItemTypes
    agent: Agent
    type: Literal["tool_call_item"]

class ToolCallOutputItem:
    raw_item: ToolCallOutputTypes
    agent: Agent
    output: Any
    type: Literal["tool_call_output_item"]

class HandoffCallItem:
    raw_item: ResponseFunctionToolCall
    agent: Agent
    type: Literal["handoff_call_item"]

class HandoffOutputItem:
    raw_item: TResponseInputItem
    agent: Agent
    source_agent: Agent
    target_agent: Agent
    type: Literal["handoff_output_item"]

class ReasoningItem:
    raw_item: ResponseReasoningItem
    agent: Agent
    type: Literal["reasoning_item"]

class ModelResponse:
    output: list[TResponseOutputItem]
    usage: Usage
    response_id: str | None

    def to_input_items() -> list[TResponseInputItem]: ...

class RawResponsesStreamEvent:
    data: TResponseStreamEvent
    type: Literal["raw_response_event"]

class RunItemStreamEvent:
    name: Literal[...]
    item: RunItem
    type: Literal["run_item_stream_event"]

class AgentUpdatedStreamEvent:
    new_agent: Agent
    type: Literal["agent_updated_stream_event"]

class ItemHelpers:
    @classmethod
    def extract_last_content(message) -> str: ...

    @classmethod
    def extract_last_text(message) -> str | None: ...

    @classmethod
    def input_to_new_input_list(input) -> list[TResponseInputItem]: ...

    @classmethod
    def text_message_outputs(items) -> str: ...

class RunResult:
    input: str | list[TResponseInputItem]
    new_items: list[RunItem]
    raw_responses: list[ModelResponse]
    final_output: Any
    input_guardrail_results: list[InputGuardrailResult]
    output_guardrail_results: list[OutputGuardrailResult]
    tool_input_guardrail_results: list[ToolInputGuardrailResult]
    tool_output_guardrail_results: list[ToolOutputGuardrailResult]
    context_wrapper: RunContextWrapper

    @property
    def last_agent() -> Agent: ...

    @property
    def last_response_id() -> str | None: ...

    def final_output_as(cls, raise_if_incorrect_type) -> T: ...
    def to_input_list() -> list[TResponseInputItem]: ...

class RunResultStreaming:
    current_agent: Agent
    current_turn: int
    max_turns: int
    is_complete: bool
    trace: Trace | None

    async def stream_events() -> AsyncIterator[StreamEvent]: ...
    def cancel(mode) -> None: ...

class AgentsException(Exception):
    run_data: RunErrorDetails | None

class MaxTurnsExceeded(AgentsException):
    message: str

class ModelBehaviorError(AgentsException):
    message: str

class UserError(AgentsException):
    message: str

class InputGuardrailTripwireTriggered(AgentsException):
    guardrail_result: InputGuardrailResult

class OutputGuardrailTripwireTriggered(AgentsException):
    guardrail_result: OutputGuardrailResult

class ToolInputGuardrailTripwireTriggered(AgentsException):
    guardrail: ToolInputGuardrail
    output: ToolGuardrailFunctionOutput

class ToolOutputGuardrailTripwireTriggered(AgentsException):
    guardrail: ToolOutputGuardrail
    output: ToolGuardrailFunctionOutput

class Usage:
    requests: int
    input_tokens: int
    input_tokens_details: InputTokensDetails
    output_tokens: int
    output_tokens_details: OutputTokensDetails
    total_tokens: int
    request_usage_entries: list[RequestUsage]

    def add(other: Usage) -> None: ...

class RunHooks[TContext]:
    async def on_llm_start(context, agent, system_prompt, input_items): ...
    async def on_llm_end(context, agent, response): ...
    async def on_agent_start(context, agent): ...
    async def on_agent_end(context, agent, output): ...
    async def on_handoff(context, from_agent, to_agent): ...
    async def on_tool_start(context, agent, tool): ...
    async def on_tool_end(context, agent, tool, result): ...

class AgentHooks[TContext]:
    async def on_start(context, agent): ...
    async def on_end(context, agent, output): ...
    async def on_handoff(context, agent, source): ...
    async def on_tool_start(context, agent, tool): ...
    async def on_tool_end(context, agent, tool, result): ...
    async def on_llm_start(context, agent, system_prompt, input_items): ...
    async def on_llm_end(context, agent, response): ...

def set_default_openai_key(key: str, use_for_tracing: bool = True) -> None:
    """
    Set default OpenAI API key for LLM requests and optionally tracing.

    Parameters:
    - key: OpenAI API key string
    - use_for_tracing: Whether to use this key for tracing (default: True)
    """

def set_default_openai_client(client: AsyncOpenAI, use_for_tracing: bool = True) -> None:
    """
    Set default OpenAI client for LLM requests.

    Parameters:
    - client: AsyncOpenAI client instance
    - use_for_tracing: Whether to use this client for tracing (default: True)
    """

def set_default_openai_api(api: Literal["chat_completions", "responses"]) -> None:
    """
    Set default API mode (responses or chat_completions).

    Parameters:
    - api: API mode to use
    """

def enable_verbose_stdout_logging() -> None:
    """Enable verbose logging to stdout for debugging."""

ApplyDiffMode = Literal["default", "create"]

def apply_diff(input: str, diff: str, mode: ApplyDiffMode) -> str:
    """
    Apply V4A diff to text.

    Parameters:
    - input: Original text
    - diff: Diff to apply
    - mode: Application mode ("default" or "create")

    Returns:
    - Modified text
    """

async def run_demo_loop(agent, *, stream, context) -> None:
    """
    Run simple REPL loop for testing agents.

    Parameters:
    - agent: Starting agent
    - stream: Whether to stream output
    - context: Context object
    """

Environment = Literal["mac", "windows", "ubuntu", "browser"]
Button = Literal["left", "right", "wheel", "back", "forward"]

class Computer:
    """Synchronous computer control interface."""

    @property
    def environment(self) -> Environment: ...

    @property
    def dimensions(self) -> tuple[int, int]: ...

    def screenshot() -> str: ...
    def click(x, y, button) -> None: ...
    def double_click(x, y) -> None: ...
    def scroll(x, y, scroll_x, scroll_y) -> None: ...
    def type(text) -> None: ...
    def wait() -> None: ...
    def move(x, y) -> None: ...
    def keypress(keys) -> None: ...
    def drag(path) -> None: ...

class AsyncComputer:
    """Asynchronous computer control interface."""

    @property
    def environment(self) -> Environment: ...

    @property
    def dimensions(self) -> tuple[int, int]: ...

    async def screenshot() -> str: ...
    async def click(x, y, button) -> None: ...
    async def double_click(x, y) -> None: ...
    async def scroll(x, y, scroll_x, scroll_y) -> None: ...
    async def type(text) -> None: ...
    async def wait() -> None: ...
    async def move(x, y) -> None: ...
    async def keypress(keys) -> None: ...
    async def drag(path) -> None: ...

__version__: str  # Package version string