from __future__ import annotations import os from collections.abc import Callable from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any, Generic, Literal from typing_extensions import NotRequired, TypedDict from .guardrail import InputGuardrail, OutputGuardrail from .handoffs import HandoffHistoryMapper, HandoffInputFilter from .items import TResponseInputItem from .lifecycle import RunHooks from .memory import Session, SessionInputCallback, SessionSettings from .model_settings import ModelSettings from .models.interface import Model, ModelProvider from .models.multi_provider import MultiProvider from .run_context import TContext from .run_error_handlers import RunErrorHandlers from .tracing import TracingConfig from .util._types import MaybeAwaitable if TYPE_CHECKING: from .agent import Agent from .run_context import RunContextWrapper from .sandbox.manifest import Manifest from .sandbox.session.base_sandbox_session import BaseSandboxSession from .sandbox.session.sandbox_client import BaseSandboxClient from .sandbox.session.sandbox_session_state import SandboxSessionState from .sandbox.snapshot import SnapshotBase, SnapshotSpec DEFAULT_MAX_TURNS = 10 DEFAULT_MAX_MANIFEST_ENTRY_CONCURRENCY = 4 DEFAULT_MAX_LOCAL_DIR_FILE_CONCURRENCY = 4 def _default_trace_include_sensitive_data() -> bool: """Return the default for trace_include_sensitive_data based on environment.""" val = os.getenv("OPENAI_AGENTS_TRACE_INCLUDE_SENSITIVE_DATA", "true") return val.strip().lower() in ("1", "true", "yes", "on") @dataclass class ModelInputData: """Container for the data that will be sent to the model.""" input: list[TResponseInputItem] instructions: str | None @dataclass class CallModelData(Generic[TContext]): """Data passed to `RunConfig.call_model_input_filter` prior to model call.""" model_data: ModelInputData agent: Agent[TContext] context: TContext | None CallModelInputFilter = Callable[[CallModelData[Any]], MaybeAwaitable[ModelInputData]] ReasoningItemIdPolicy = Literal["preserve", "omit"] @dataclass class ToolErrorFormatterArgs(Generic[TContext]): """Data passed to ``RunConfig.tool_error_formatter`` callbacks.""" kind: Literal["approval_rejected"] """The category of tool error being formatted.""" tool_type: Literal["function", "computer", "shell", "apply_patch", "custom"] """The tool runtime that produced the error.""" tool_name: str """The name of the tool that produced the error.""" call_id: str """The unique tool call identifier.""" default_message: str """The SDK default message for this error kind.""" run_context: RunContextWrapper[TContext] """The active run context for the current execution.""" ToolErrorFormatter = Callable[[ToolErrorFormatterArgs[Any]], MaybeAwaitable[str | None]] @dataclass class SandboxConcurrencyLimits: """Concurrency limits for sandbox materialization work.""" manifest_entries: int | None = DEFAULT_MAX_MANIFEST_ENTRY_CONCURRENCY """Maximum number of manifest entries to materialize concurrently per sandbox session. Set to `None` to disable this manifest entry limit. """ local_dir_files: int | None = DEFAULT_MAX_LOCAL_DIR_FILE_CONCURRENCY """Maximum number of files to copy concurrently for each local_dir manifest entry. Set to `None` to disable this per-local-dir file copy limit. """ def validate(self) -> None: if self.manifest_entries is not None and self.manifest_entries < 1: raise ValueError("concurrency_limits.manifest_entries must be at least 1") if self.local_dir_files is not None and self.local_dir_files < 1: raise ValueError("concurrency_limits.local_dir_files must be at least 1") @dataclass class SandboxRunConfig: """Grouped sandbox runtime configuration for `Runner`.""" client: BaseSandboxClient[Any] | None = None """Sandbox client used to create or resume sandbox sessions.""" options: Any | None = None """Sandbox-client-specific options used when creating a fresh session.""" session: BaseSandboxSession | None = None """Live sandbox session override for the current process.""" session_state: SandboxSessionState | None = None """Explicit sandbox session state to resume from when not using `RunState` payloads.""" manifest: Manifest | None = None """Optional sandbox manifest override for fresh session creation.""" snapshot: SnapshotSpec | SnapshotBase | None = None """Optional sandbox snapshot used for fresh session creation.""" concurrency_limits: SandboxConcurrencyLimits = field(default_factory=SandboxConcurrencyLimits) """Concurrency limits for sandbox materialization work.""" @dataclass class RunConfig: """Configures settings for the entire agent run.""" model: str | Model | None = None """The model to use for the entire agent run. If set, will override the model set on every agent. The model_provider passed in below must be able to resolve this model name. """ model_provider: ModelProvider = field(default_factory=MultiProvider) """The model provider to use when looking up string model names. Defaults to OpenAI.""" model_settings: ModelSettings | None = None """Configure global model settings. Any non-null values will override the agent-specific model settings. """ handoff_input_filter: HandoffInputFilter | None = None """A global input filter to apply to all handoffs. If `Handoff.input_filter` is set, then that will take precedence. The input filter allows you to edit the inputs that are sent to the new agent. See the documentation in `Handoff.input_filter` for more details. Server-managed conversations (`conversation_id`, `previous_response_id`, or `auto_previous_response_id`) do not support handoff input filters. """ nest_handoff_history: bool = False """Opt-in beta: wrap prior run history in a single assistant message before handing off when no custom input filter is set. This is disabled by default while we stabilize nested handoffs; set to True to enable the collapsed transcript behavior. Server-managed conversations (`conversation_id`, `previous_response_id`, or `auto_previous_response_id`) automatically disable this behavior with a warning. """ handoff_history_mapper: HandoffHistoryMapper | None = None """Optional function that receives the normalized transcript (history + handoff items) and returns the input history that should be passed to the next agent. When left as `None`, the runner collapses the transcript into a single assistant message. This function only runs when `nest_handoff_history` is True. """ input_guardrails: list[InputGuardrail[Any]] | None = None """A list of input guardrails to run on the initial run input.""" output_guardrails: list[OutputGuardrail[Any]] | None = None """A list of output guardrails to run on the final output of the run.""" tracing_disabled: bool = False """Whether tracing is disabled for the agent run. If disabled, we will not trace the agent run. """ tracing: TracingConfig | None = None """Tracing configuration for this run.""" trace_include_sensitive_data: bool = field( default_factory=_default_trace_include_sensitive_data ) """Whether we include potentially sensitive data (for example: inputs/outputs of tool calls or LLM generations) in traces. If False, we'll still create spans for these events, but the sensitive data will not be included. """ workflow_name: str = "Agent workflow" """The name of the run, used for tracing. Should be a logical name for the run, like "Code generation workflow" or "Customer support agent". """ trace_id: str | None = None """A custom trace ID to use for tracing. If not provided, we will generate a new trace ID.""" group_id: str | None = None """ A grouping identifier to use for tracing, to link multiple traces from the same conversation or process. For example, you might use a chat thread ID. """ trace_metadata: dict[str, Any] | None = None """ An optional dictionary of additional metadata to include with the trace. """ session_input_callback: SessionInputCallback | None = None """Defines how to handle session history when new input is provided. - `None` (default): The new input is appended to the session history. - `SessionInputCallback`: A custom function that receives the history and new input, and returns the desired combined list of items. """ call_model_input_filter: CallModelInputFilter | None = None """ Optional callback that is invoked immediately before calling the model. It receives the current agent, context and the model input (instructions and input items), and must return a possibly modified `ModelInputData` to use for the model call. This allows you to edit the input sent to the model e.g. to stay within a token limit. For example, you can use this to add a system prompt to the input. """ tool_error_formatter: ToolErrorFormatter | None = None """Optional callback that formats tool error messages returned to the model. Returning ``None`` falls back to the SDK default message. """ session_settings: SessionSettings | None = None """Configure session settings. Any non-null values will override the session's default settings. Used to control session behavior like the number of items to retrieve. """ reasoning_item_id_policy: ReasoningItemIdPolicy | None = None """Controls how reasoning items are converted to next-turn model input. - ``None`` / ``"preserve"`` keeps reasoning item IDs as-is. - ``"omit"`` strips reasoning item IDs from model input built by the runner. """ sandbox: SandboxRunConfig | None = None """Optional sandbox runtime configuration for `SandboxAgent` execution.""" class RunOptions(TypedDict, Generic[TContext]): """Arguments for ``AgentRunner`` methods.""" context: NotRequired[TContext | None] """The context for the run.""" max_turns: NotRequired[int] """The maximum number of turns to run for.""" hooks: NotRequired[RunHooks[TContext] | None] """Lifecycle hooks for the run.""" run_config: NotRequired[RunConfig | None] """Run configuration.""" previous_response_id: NotRequired[str | None] """The ID of the previous response, if any.""" auto_previous_response_id: NotRequired[bool] """Enable automatic response chaining for the first turn.""" conversation_id: NotRequired[str | None] """The ID of the stored conversation, if any.""" session: NotRequired[Session | None] """The session for the run.""" error_handlers: NotRequired[RunErrorHandlers[TContext] | None] """Error handlers keyed by error kind. Currently supports max_turns.""" __all__ = [ "DEFAULT_MAX_TURNS", "CallModelData", "CallModelInputFilter", "ModelInputData", "ReasoningItemIdPolicy", "RunConfig", "RunOptions", "SandboxConcurrencyLimits", "SandboxRunConfig", "ToolErrorFormatter", "ToolErrorFormatterArgs", "_default_trace_include_sensitive_data", ]