`Sandbox`

SandboxAgent `dataclass`

Bases: Agent[TContext]

An Agent with sandbox-specific configuration.

Runtime transport details such as the sandbox client, client options, and live session are provided at run time through RunConfig(sandbox=...), not stored on the agent itself.

Source code in src/agents/sandbox/sandbox_agent.py

@dataclass
class SandboxAgent(Agent[TContext]):
    """An `Agent` with sandbox-specific configuration.

    Runtime transport details such as the sandbox client, client options, and live session are
    provided at run time through `RunConfig(sandbox=...)`, not stored on the agent itself.
    """

    default_manifest: Manifest | None = None
    """Default sandbox manifest for new sessions created by `Runner` sandbox execution."""

    base_instructions: (
        str
        | Callable[
            [RunContextWrapper[TContext], Agent[TContext]], Awaitable[str | None] | str | None
        ]
        | None
    ) = None
    """Override for the SDK sandbox base prompt. Most callers should use `instructions`."""

    capabilities: Sequence[Capability] = field(default_factory=Capabilities.default)
    """Sandbox capabilities that can mutate the manifest, add instructions, and expose tools."""

    run_as: User | str | None = None
    """User identity used for model-facing sandbox tools such as shell, file reads, and patches."""

    _sandbox_concurrency_guard: object | None = field(default=None, init=False, repr=False)

    if TYPE_CHECKING:

        def __init__(
            self,
            name: str,
            handoff_description: str | None = None,
            tools: list[Tool] = ...,
            mcp_servers: list[MCPServer] = ...,
            mcp_config: MCPConfig = ...,
            instructions: (
                str
                | Callable[
                    [RunContextWrapper[TContext], Agent[TContext]],
                    MaybeAwaitable[str],
                ]
                | None
            ) = None,
            prompt: Prompt | DynamicPromptFunction | None = None,
            handoffs: list[Agent[Any] | Handoff[TContext, Any]] = ...,
            model: str | Model | None = None,
            model_settings: ModelSettings | dict[str, Any] = ...,
            input_guardrails: list[InputGuardrail[TContext]] = ...,
            output_guardrails: list[OutputGuardrail[TContext]] = ...,
            output_type: type[Any] | AgentOutputSchemaBase | None = None,
            hooks: AgentHooks[TContext] | None = None,
            tool_use_behavior: (
                Literal["run_llm_again", "stop_on_first_tool"]
                | StopAtTools
                | ToolsToFinalOutputFunction
            ) = "run_llm_again",
            reset_tool_choice: bool = True,
            default_manifest: Manifest | dict[str, Any] | None = None,
            base_instructions: (
                str
                | Callable[
                    [RunContextWrapper[TContext], Agent[TContext]],
                    Awaitable[str | None] | str | None,
                ]
                | None
            ) = None,
            capabilities: Sequence[Capability] = ...,
            run_as: User | dict[str, Any] | str | None = None,
        ) -> None: ...

    def __post_init__(self) -> None:
        super().__post_init__()
        if isinstance(self.default_manifest, dict):
            self.default_manifest = _coerce_manifest(
                self.default_manifest, parameter_name="sandbox.default_manifest"
            )
        if isinstance(self.run_as, dict):
            self.run_as = coerce_pydantic_config(self.run_as, User, parameter_name="sandbox.run_as")
        if (
            self.base_instructions is not None
            and not isinstance(self.base_instructions, str)
            and not callable(self.base_instructions)
        ):
            raise TypeError(
                f"SandboxAgent base_instructions must be a string, callable, or None, "
                f"got {type(self.base_instructions).__name__}"
            )
        if self.run_as is not None and not isinstance(self.run_as, str | User):
            raise TypeError(
                f"SandboxAgent run_as must be a string, User, or None, "
                f"got {type(self.run_as).__name__}"
            )

name `instance-attribute`

name: str

The name of the agent.

handoff_description `class-attribute` `instance-attribute`

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 `class-attribute` `instance-attribute`

tools: list[Tool] = field(default_factory=list)

A list of tools that the agent can use.

mcp_servers `class-attribute` `instance-attribute`

mcp_servers: list[MCPServer] = field(default_factory=list)

A list of Model Context Protocol 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. Consider using MCPServerManager from agents.mcp to keep connect/cleanup in the same task.

mcp_config `class-attribute` `instance-attribute`

mcp_config: MCPConfig = field(
    default_factory=lambda: MCPConfig()
)

Configuration for MCP servers.

instructions `class-attribute` `instance-attribute`

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 `class-attribute` `instance-attribute`

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 `class-attribute` `instance-attribute`

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 `class-attribute` `instance-attribute`

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-5.6-luna").

model_settings `class-attribute` `instance-attribute`

model_settings: ModelSettings = field(
    default_factory=get_default_model_settings
)

Configures model-specific tuning parameters (e.g. temperature, top_p).

Accepts a ModelSettings instance or a dictionary containing its fields.

input_guardrails `class-attribute` `instance-attribute`

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 `class-attribute` `instance-attribute`

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 `class-attribute` `instance-attribute`

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 `class-attribute` `instance-attribute`

hooks: AgentHooks[TContext] | None = None

A class that receives callbacks on various lifecycle events for this agent.

tool_use_behavior `class-attribute` `instance-attribute`

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 `class-attribute` `instance-attribute`

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.

default_manifest `class-attribute` `instance-attribute`

default_manifest: Manifest | None = None

Default sandbox manifest for new sessions created by Runner sandbox execution.

base_instructions `class-attribute` `instance-attribute`

base_instructions: (
    str
    | Callable[
        [RunContextWrapper[TContext], Agent[TContext]],
        Awaitable[str | None] | str | None,
    ]
    | None
) = None

Override for the SDK sandbox base prompt. Most callers should use instructions.

capabilities `class-attribute` `instance-attribute`

capabilities: Sequence[Capability] = field(
    default_factory=default
)

Sandbox capabilities that can mutate the manifest, add instructions, and expose tools.

run_as `class-attribute` `instance-attribute`

run_as: User | str | None = None

User identity used for model-facing sandbox tools such as shell, file reads, and patches.

get_mcp_tools `async`

get_mcp_tools(
    run_context: RunContextWrapper[TContext],
) -> list[Tool]

Fetches the available tools from the MCP servers.

Source code in src/agents/agent.py

async def get_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)
    failure_error_function = self.mcp_config.get(
        "failure_error_function", default_tool_error_function
    )
    include_server_in_tool_names = self.mcp_config.get("include_server_in_tool_names", False)
    reserved_tool_names = (
        await self._get_mcp_tool_reserved_names(run_context)
        if include_server_in_tool_names
        else None
    )
    return await MCPUtil.get_all_function_tools(
        self.mcp_servers,
        convert_schemas_to_strict,
        run_context,
        self,
        failure_error_function=failure_error_function,
        include_server_in_tool_names=include_server_in_tool_names,
        reserved_tool_names=reserved_tool_names,
    )

get_all_tools `async`

get_all_tools(
    run_context: RunContextWrapper[TContext],
) -> list[Tool]

All agent tools, including MCP tools and function tools.

Source code in src/agents/agent.py

async def get_all_tools(self, run_context: RunContextWrapper[TContext]) -> list[Tool]:
    """All agent tools, including MCP tools and function tools."""
    mcp_tools = await self.get_mcp_tools(run_context)

    async def _check_tool_enabled(tool: Tool) -> bool:
        if not isinstance(tool, FunctionTool):
            return True

        attr = tool.is_enabled
        if isinstance(attr, bool):
            return attr
        res = attr(run_context, self)
        if inspect.isawaitable(res):
            return bool(await res)
        return bool(res)

    results = await gather_with_cancel(*(_check_tool_enabled(t) for t in self.tools))
    enabled: list[Tool] = [t for t, ok in zip(self.tools, results, strict=False) if ok]
    all_tools: list[Tool] = prune_orphaned_tool_search_tools([*mcp_tools, *enabled])
    _validate_codex_tool_name_collisions(all_tools)
    return all_tools

clone

clone(**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:

new_agent = agent.clone(instructions="New instructions")

Source code in src/agents/agent.py

def clone(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")
        ```
    """
    if (
        "model" in kwargs
        and "model_settings" not in kwargs
        and _model_settings_match_implicit_model_defaults(self.model, self.model_settings)
    ):
        kwargs["model_settings"] = _initial_model_settings_for_model(kwargs["model"])
    if "model_settings" in kwargs:
        kwargs["model_settings"] = _coerce_model_settings(
            kwargs["model_settings"],
            parameter_name="Agent model_settings",
            model_settings_type=type(self.model_settings),
            inherited_model_settings=self.model_settings,
        )
    return dataclasses.replace(self, **kwargs)

as_tool

as_tool(
    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 | dict[str, Any] | 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,
    needs_approval: bool
    | Callable[
        [RunContextWrapper[Any], dict[str, Any], str],
        Awaitable[bool],
    ] = False,
    parameters: type[Any] | None = None,
    input_builder: StructuredToolInputBuilder | None = None,
    include_input_schema: bool = False,
) -> FunctionTool

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.

Parameters:

Name	Type	Description	Default
`tool_name`	`str \| None`	The name of the tool. If not provided, the agent's name will be used.	required
`tool_description`	`str \| None`	The description of the tool, which should indicate what it does and when to use it.	required
`custom_output_extractor`	`Callable[[RunResult \| RunResultStreaming], Awaitable[str]] \| None`	A function that extracts the output from the agent. If not provided, the last message from the agent will be used. Nested run results expose `agent_tool_invocation` metadata when this agent is invoked via `as_tool()`.	`None`
`is_enabled`	`bool \| Callable[[RunContextWrapper[Any], AgentBase[Any]], MaybeAwaitable[bool]]`	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.	`True`
`on_stream`	`Callable[[AgentToolStreamEvent], MaybeAwaitable[None]] \| None`	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.	`None`
`failure_error_function`	`ToolErrorFunction \| None`	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.	`default_tool_error_function`
`needs_approval`	`bool \| Callable[[RunContextWrapper[Any], dict[str, Any], str], Awaitable[bool]]`	Bool or callable to decide if this agent tool should pause for approval.	`False`
`parameters`	`type[Any] \| None`	Structured input type for the tool arguments (dataclass or Pydantic model).	`None`
`input_builder`	`StructuredToolInputBuilder \| None`	Optional function to build the nested agent input from structured data.	`None`
`include_input_schema`	`bool`	Whether to include the full JSON schema in structured input.	`False`

Source code in src/agents/agent.py

def as_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 | dict[str, Any] | 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,
    needs_approval: bool
    | Callable[[RunContextWrapper[Any], dict[str, Any], str], Awaitable[bool]] = False,
    parameters: type[Any] | None = None,
    input_builder: StructuredToolInputBuilder | None = None,
    include_input_schema: bool = False,
) -> FunctionTool:
    """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. Nested run results expose
            `agent_tool_invocation` metadata when this agent is invoked via `as_tool()`.
        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.
        needs_approval: Bool or callable to decide if this agent tool should pause for approval.
        parameters: Structured input type for the tool arguments (dataclass or Pydantic model).
        input_builder: Optional function to build the nested agent input from structured data.
        include_input_schema: Whether to include the full JSON schema in structured input.
    """

    if run_config is not None:
        from .run_config import _coerce_run_config

        run_config = _coerce_run_config(run_config)

    def _is_supported_parameters(value: Any) -> bool:
        if not isinstance(value, type):
            return False
        if dataclasses.is_dataclass(value):
            return True
        return issubclass(value, BaseModel)

    tool_name_resolved = tool_name or _transforms.transform_string_function_style(self.name)
    tool_description_resolved = tool_description or ""
    has_custom_parameters = parameters is not None
    include_schema = bool(include_input_schema and has_custom_parameters)
    should_capture_tool_input = bool(
        has_custom_parameters or include_schema or input_builder is not None
    )

    if parameters is None:
        params_adapter = TypeAdapter(AgentAsToolInput)
        params_schema = ensure_strict_json_schema(params_adapter.json_schema())
    else:
        if not _is_supported_parameters(parameters):
            raise TypeError("Agent tool parameters must be a dataclass or Pydantic model type.")
        params_adapter = TypeAdapter(parameters)
        params_schema = ensure_strict_json_schema(params_adapter.json_schema())

    schema_info = build_structured_input_schema_info(
        params_schema,
        include_json_schema=include_schema,
    )

    def _normalize_tool_input(parsed: Any, tool_name: str) -> Any:
        # Prefer JSON mode so structured params (datetime/UUID/Decimal, etc.) serialize cleanly.
        try:
            return params_adapter.dump_python(parsed, mode="json")
        except Exception as exc:
            raise ModelBehaviorError(
                f"Failed to serialize structured tool input for {tool_name}: {exc}"
            ) from exc

    async def _run_agent_impl(context: ToolContext, input_json: str) -> Any:
        from .run import DEFAULT_MAX_TURNS, Runner
        from .tool_context import ToolContext

        tool_name = (
            context.tool_name if isinstance(context, ToolContext) else tool_name_resolved
        )
        json_data = _parse_function_tool_json_input(
            tool_name=tool_name,
            input_json=input_json,
        )
        _log_function_tool_invocation(tool_name=tool_name, input_json=input_json)

        base_message = f"Invalid JSON input for tool {tool_name}"
        validation_failed = False
        try:
            parsed_params = params_adapter.validate_python(json_data)
        except ValidationError as exc:
            if not _debug.DONT_LOG_TOOL_DATA:
                raise ModelBehaviorError(f"{base_message}: {exc}") from exc
            validation_failed = True

        if validation_failed:
            raise ModelBehaviorError(base_message)

        params_data = _normalize_tool_input(parsed_params, tool_name)
        resolved_input = await resolve_agent_tool_input(
            params=params_data,
            schema_info=schema_info if should_capture_tool_input else None,
            input_builder=input_builder,
        )
        if not isinstance(resolved_input, str) and not isinstance(resolved_input, list):
            raise ModelBehaviorError("Agent tool called with invalid input")

        resolved_max_turns = max_turns if max_turns is not None else DEFAULT_MAX_TURNS
        resolved_run_config = run_config
        if resolved_run_config is None and isinstance(context, ToolContext):
            resolved_run_config = context.run_config
        tool_state_scope_id = get_agent_tool_state_scope(context)
        if isinstance(context, ToolContext):
            # Use a fresh ToolContext to avoid sharing approval state with parent runs.
            nested_context = ToolContext(
                context=context.context,
                usage=context.usage,
                tool_name=context.tool_name,
                tool_call_id=context.tool_call_id,
                tool_arguments=context.tool_arguments,
                tool_call=context.tool_call,
                tool_namespace=context.tool_namespace,
                agent=context.agent,
                run_config=resolved_run_config,
            )
            set_agent_tool_state_scope(nested_context, tool_state_scope_id)
            if should_capture_tool_input:
                nested_context.tool_input = params_data
        elif isinstance(context, RunContextWrapper):
            if should_capture_tool_input:
                nested_context = RunContextWrapper(context=context.context)
                set_agent_tool_state_scope(nested_context, tool_state_scope_id)
                nested_context.tool_input = params_data
            else:
                nested_context = context.context
        else:
            if should_capture_tool_input:
                nested_context = RunContextWrapper(context=context)
                set_agent_tool_state_scope(nested_context, tool_state_scope_id)
                nested_context.tool_input = params_data
            else:
                nested_context = context
        run_result: RunResult | RunResultStreaming | None = None
        resume_state: RunState | None = None
        should_record_run_result = True

        def _nested_approvals_status(
            pending_run_result: RunResult | RunResultStreaming,
        ) -> Literal["approved", "pending", "rejected"]:
            interruptions = pending_run_result.interruptions
            nested_decision_context = pending_run_result.to_state()._context
            has_pending = False
            has_decision = False
            for interruption in interruptions:
                call_id = get_tool_approval_item_call_id(interruption)
                if not call_id:
                    has_pending = True
                    continue
                tool_namespace = RunContextWrapper._resolve_tool_namespace(interruption)
                status = (
                    nested_decision_context.get_approval_status(
                        interruption.tool_name or "",
                        call_id,
                        tool_namespace=tool_namespace,
                        existing_pending=interruption,
                    )
                    if nested_decision_context is not None
                    else None
                )
                if (
                    status is None
                    and nested_decision_context is not None
                    and context._allow_legacy_approval_binding_reconstruction
                ):
                    status = context.get_approval_status(
                        interruption.tool_name or "",
                        call_id,
                        tool_namespace=tool_namespace,
                        existing_pending=interruption,
                    )
                    if status is not None:
                        legacy_namespace = RunContextWrapper._resolve_tool_namespace(
                            interruption
                        )
                        legacy_tool_name = RunContextWrapper._resolve_tool_name(interruption)
                        legacy_qualified_key = (
                            f"{legacy_namespace}.{legacy_tool_name}"
                            if legacy_namespace is not None
                            else legacy_tool_name
                        )
                        approval_keys = (
                            RunContextWrapper._resolve_approval_key(interruption),
                            *RunContextWrapper._resolve_approval_keys(interruption),
                            legacy_qualified_key,
                        )
                        approval_record = next(
                            (
                                context._approvals[key]
                                for key in approval_keys
                                if key in context._approvals
                            ),
                            None,
                        )
                        if status:
                            RunContextWrapper.approve_tool(
                                nested_decision_context,
                                interruption,
                                always_approve=bool(
                                    approval_record and approval_record.approved is True
                                ),
                            )
                        else:
                            RunContextWrapper.reject_tool(
                                nested_decision_context,
                                interruption,
                                always_reject=bool(
                                    approval_record and approval_record.rejected is True
                                ),
                                rejection_message=context.get_rejection_message(
                                    interruption.tool_name or "",
                                    call_id,
                                    tool_namespace=tool_namespace,
                                    existing_pending=interruption,
                                ),
                            )
                if status is False:
                    return "rejected"
                if status is True:
                    has_decision = True
                if status is None:
                    has_pending = True
            if has_decision:
                return "approved"
            if has_pending:
                return "pending"
            return "approved"

        if isinstance(context, ToolContext) and context.tool_call is not None:
            pending_run_result = peek_agent_tool_run_result(
                context.tool_call,
                scope_id=tool_state_scope_id,
            )
            pending_resume_state = get_agent_tool_resume_state(pending_run_result)
            if pending_resume_state is not None:
                resume_state = pending_resume_state
            elif pending_run_result and getattr(pending_run_result, "interruptions", None):
                resolved_pending_result = cast(
                    "RunResult | RunResultStreaming",
                    pending_run_result,
                )
                status = _nested_approvals_status(resolved_pending_result)
                if status == "pending":
                    run_result = resolved_pending_result
                    should_record_run_result = False
                elif status in ("approved", "rejected"):
                    resume_state = resolved_pending_result.to_state()
                    if resume_state._context is not None:
                        # Keep accumulating nested post-resume usage on the parent
                        # ToolContext accumulator. resolve_resumed_context only
                        # replaces application .context and would otherwise leave
                        # the restored nested wrapper on a detached Usage object.
                        resume_state._context.usage = context.usage
                    record_agent_tool_resume_state(
                        context.tool_call,
                        resume_state,
                        scope_id=tool_state_scope_id,
                        approval_items=resolved_pending_result.interruptions,
                    )

        if run_result is None:
            if on_stream is not None:
                stream_handler = on_stream
                run_result_streaming = Runner.run_streamed(
                    starting_agent=cast(Agent[Any], self),
                    input=resume_state if resume_state is not None else resolved_input,
                    # On resume, pass the parent application context so
                    # resolve_resumed_context can update the nested restored
                    # wrapper's .context without dropping nested approvals.
                    context=cast(Any, nested_context),
                    run_config=resolved_run_config,
                    max_turns=resolved_max_turns,
                    hooks=hooks,
                    previous_response_id=None
                    if resume_state is not None
                    else previous_response_id,
                    conversation_id=None if resume_state is not None else 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()

                async def _run_handler(payload: AgentToolStreamEvent) -> None:
                    """Execute the user callback while capturing exceptions."""
                    try:
                        maybe_result = stream_handler(payload)
                        if inspect.isawaitable(maybe_result):
                            await maybe_result
                    except Exception as exc:

                        def diagnostic_extra() -> dict[str, object]:
                            return {"agent_name": self.name}

                        log_model_and_tool_action_error(
                            logger,
                            "Error while handling an agent tool on_stream event",
                            exc,
                            diagnostic_extra=diagnostic_extra,
                        )

                async def dispatch_stream_events() -> None:
                    while True:
                        payload = await event_queue.get()
                        is_sentinel = payload is None  # None marks the end of the stream.
                        try:
                            if payload is not None:
                                await _run_handler(payload)
                        finally:
                            event_queue.task_done()

                        if is_sentinel:
                            break

                async def enqueue_stream_events() -> None:
                    from .stream_events import AgentUpdatedStreamEvent

                    current_agent = run_result_streaming.current_agent
                    try:
                        async for event in run_result_streaming.stream_events():
                            if isinstance(event, AgentUpdatedStreamEvent):
                                current_agent = event.new_agent

                            payload: AgentToolStreamEvent = {
                                "event": event,
                                "agent": current_agent,
                                "tool_call": context.tool_call,
                            }
                            await event_queue.put(payload)
                    finally:
                        await event_queue.put(None)

                await run_producer_consumer(enqueue_stream_events(), dispatch_stream_events())
                run_result = run_result_streaming
            else:
                run_result = await Runner.run(
                    starting_agent=cast(Agent[Any], self),
                    input=resume_state if resume_state is not None else resolved_input,
                    # On resume, pass the parent application context so
                    # resolve_resumed_context can update the nested restored
                    # wrapper's .context without dropping nested approvals.
                    context=cast(Any, nested_context),
                    run_config=resolved_run_config,
                    max_turns=resolved_max_turns,
                    hooks=hooks,
                    previous_response_id=None
                    if resume_state is not None
                    else previous_response_id,
                    conversation_id=None if resume_state is not None else conversation_id,
                    session=session,
                )
        assert run_result is not None

        # Store the run result by tool call identity so nested interruptions can be read later.
        interruptions = getattr(run_result, "interruptions", None)
        if isinstance(context, ToolContext) and context.tool_call is not None and interruptions:
            if should_record_run_result:
                record_agent_tool_run_result(
                    context.tool_call,
                    run_result,
                    scope_id=tool_state_scope_id,
                )
            return run_result.final_output

        if custom_output_extractor is not None:
            return await custom_output_extractor(run_result)

        if run_result.final_output is not None and (
            not isinstance(run_result.final_output, str) or run_result.final_output != ""
        ):
            return run_result.final_output

        from .items import ItemHelpers, MessageOutputItem, ToolCallOutputItem

        for item in reversed(run_result.new_items):
            if isinstance(item, MessageOutputItem):
                text_output = ItemHelpers.text_message_output(item)
                if text_output:
                    return text_output

            if (
                isinstance(item, ToolCallOutputItem)
                and isinstance(item.output, str)
                and item.output
            ):
                return item.output

        return run_result.final_output

    run_agent_tool = _build_wrapped_function_tool(
        name=tool_name_resolved,
        description=tool_description_resolved,
        params_json_schema=params_schema,
        invoke_tool_impl=_run_agent_impl,
        on_handled_error=_build_handled_function_tool_error_handler(
            span_message="Error running tool (non-fatal)",
            span_message_for_json_decode_error="Error running tool",
            log_label="Tool",
        ),
        failure_error_function=failure_error_function,
        strict_json_schema=True,
        is_enabled=is_enabled,
        needs_approval=needs_approval,
        tool_origin=ToolOrigin(
            type=ToolOriginType.AGENT_AS_TOOL,
            agent_name=self.name,
            agent_tool_name=tool_name_resolved,
        ),
    )
    run_agent_tool._is_agent_tool = True
    if not tool_name:
        run_agent_tool._agent_tool_default_identity = (self.name, tool_name_resolved)
    run_agent_tool._agent_instance = self

    return run_agent_tool

get_prompt `async`

get_prompt(
    run_context: RunContextWrapper[TContext],
) -> ResponsePromptParam | None

Get the prompt for the agent.

Source code in src/agents/agent.py

async def get_prompt(
    self, run_context: RunContextWrapper[TContext]
) -> ResponsePromptParam | None:
    """Get the prompt for the agent."""
    from ._public_agent import get_public_agent

    return await PromptUtil.to_model_input(
        self.prompt,
        run_context,
        cast(Agent[TContext], get_public_agent(self)),
    )

Manifest

Bases: BaseModel

Source code in src/agents/sandbox/manifest.py

class Manifest(BaseModel):
    version: Literal[1] = 1
    root: str = Field(default="/workspace")
    entries: dict[str | Path, BaseEntry] = Field(default_factory=dict)
    environment: Environment = Field(default_factory=Environment)
    users: list[User] = Field(default_factory=list)
    groups: list[Group] = Field(default_factory=list)
    extra_path_grants: tuple[SandboxPathGrant, ...] = Field(default_factory=tuple)
    remote_mount_command_allowlist: list[str] = Field(
        default_factory=lambda: list(DEFAULT_REMOTE_MOUNT_COMMAND_ALLOWLIST)
    )
    _mount_credential_exposure_policy: _MountCredentialExposurePolicy = PrivateAttr(
        default_factory=_MountCredentialExposurePolicy
    )

    @model_validator(mode="before")
    @classmethod
    def _reject_mount_credential_exposure_policy_input(cls, value: object) -> object:
        if isinstance(value, Mapping) and _MOUNT_CREDENTIAL_EXPOSURE_POLICY_KEYS.intersection(
            value
        ):
            raise TypeError(
                "In-container mount credential exposure must be configured on a trusted "
                "Manifest instance, not in manifest input."
            )
        return value

    @field_validator("entries", mode="before")
    @classmethod
    def _parse_entries(cls, value: object) -> dict[str | Path, BaseEntry]:
        if value is None:
            return {}
        if not isinstance(value, Mapping):
            raise TypeError(f"Artifact mapping must be a mapping, got {type(value).__name__}")
        return {key: BaseEntry.parse(entry) for key, entry in value.items()}

    @field_serializer("entries", when_used="json")
    def _serialize_entries(self, entries: Mapping[str | Path, BaseEntry]) -> dict[str, object]:
        out: dict[str, object] = {}
        for key, entry in entries.items():
            key_str = key.as_posix() if isinstance(key, Path) else str(key)
            out[key_str] = entry.model_dump(mode="json")
        return out

    def validated_entries(self) -> dict[str | Path, BaseEntry]:
        validated: dict[str | Path, BaseEntry] = dict(self.entries)
        for _path, _artifact in self.iter_entries():
            pass
        return validated

    @redact_mount_validation_error_data_sync
    def with_in_container_mount_credential_exposure_acknowledged(
        self, *mount_paths: str | PurePath
    ) -> "Manifest":
        """Acknowledge mount-scoped credential exposure for exact in-container mount paths.

        This trusted application-side policy is runtime-only and is not serialized.
        """

        return self._with_mount_credential_exposure_acknowledged(
            "mount_scoped",
            mount_paths,
        )

    @redact_mount_validation_error_data_sync
    def with_in_container_mount_broad_credential_exposure_acknowledged(
        self, *mount_paths: str | PurePath
    ) -> "Manifest":
        """Acknowledge broad credential exposure for exact in-container mount paths.

        Broad authority includes managed or workload identity and external credential files.
        This trusted application-side policy is runtime-only and is not serialized.
        """

        return self._with_mount_credential_exposure_acknowledged(
            "broad",
            mount_paths,
        )

    def _with_mount_credential_exposure_acknowledged(
        self,
        authority: Literal["mount_scoped", "broad"],
        mount_paths: tuple[str | PurePath, ...],
    ) -> "Manifest":
        if not mount_paths:
            raise TypeError("At least one in-container mount path is required.")

        acknowledged: set[str] = set()
        for path in mount_paths:
            key = self._mount_credential_exposure_policy_key(path, reject_root=True)
            assert key is not None
            acknowledged.add(key)
        from ._mount_security import _validate_manifest_mount_provenance

        _validate_manifest_mount_provenance(self)
        trusted = self.model_copy(deep=True)
        current = self._mount_credential_exposure_policy
        trusted._mount_credential_exposure_policy = _MountCredentialExposurePolicy(
            mount_scoped=(
                current.mount_scoped | acknowledged
                if authority == "mount_scoped"
                else current.mount_scoped
            ),
            broad=(current.broad | acknowledged if authority == "broad" else current.broad),
        )
        return trusted

    def _acknowledges_in_container_mount_credential_exposure(
        self,
        mount_path: str | PurePath,
        authority: Literal["mount_scoped", "broad"],
    ) -> bool:
        key = self._mount_credential_exposure_policy_key(mount_path, reject_root=False)
        if key is None:
            return False
        lookup_keys = {key}
        kind, _, path_text = key.partition(":")
        root = coerce_posix_path(self.root)
        root_normalized = PurePosixPath(
            "/",
            *[part for part in root.parts if part not in {"/", ""}],
        )
        if kind == "absolute":
            try:
                relative = PurePosixPath(path_text).relative_to(root_normalized)
            except ValueError:
                pass
            else:
                if relative.parts:
                    lookup_keys.add(f"relative:{relative.as_posix()}")
        else:
            absolute = root_normalized / PurePosixPath(path_text)
            lookup_keys.add(f"absolute:{absolute.as_posix()}")
        acknowledged = getattr(self._mount_credential_exposure_policy, authority)
        return not lookup_keys.isdisjoint(acknowledged)

    def _copy_mount_credential_exposure_policy_from(self, *sources: "Manifest") -> None:
        mount_scoped: set[str] = set()
        broad: set[str] = set()
        for source in sources:
            mount_scoped.update(source._mount_credential_exposure_policy.mount_scoped)
            broad.update(source._mount_credential_exposure_policy.broad)
        self._mount_credential_exposure_policy = _MountCredentialExposurePolicy(
            mount_scoped=frozenset(mount_scoped),
            broad=frozenset(broad),
        )

    def _merge_mount_credential_exposure_policy(
        self,
        policy: _MountCredentialExposurePolicy,
    ) -> _MountCredentialExposurePolicy:
        current = self._mount_credential_exposure_policy
        merged = _MountCredentialExposurePolicy(
            mount_scoped=current.mount_scoped | policy.mount_scoped,
            broad=current.broad | policy.broad,
        )
        self._mount_credential_exposure_policy = merged
        return merged

    def _mount_credential_exposure_policy_key(
        self,
        value: str | PurePath,
        *,
        reject_root: bool,
    ) -> str | None:
        text = value.as_posix() if isinstance(value, PurePath) else value
        if not text:
            if reject_root:
                raise ValueError("Mount credential exposure path must identify a non-root path.")
            return None
        if "\\" in text:
            raise ValueError("Mount credential exposure paths must use '/' separators.")
        if reject_root and any(character in text for character in "*?[]"):
            raise ValueError("Mount credential exposure paths must not contain wildcard syntax.")

        raw = PurePosixPath(text)
        if reject_root and ".." in raw.parts:
            raise ValueError("Mount credential exposure paths must not contain parent segments.")
        if not raw.is_absolute():
            rel = self._normalize_rel_path_within_root(
                posix_path_as_path(raw),
                original=posix_path_as_path(raw),
            )
            if not rel.parts:
                if reject_root:
                    raise ValueError(
                        "Mount credential exposure path must identify a non-root path."
                    )
                return None
            return f"relative:{coerce_posix_path(rel).as_posix()}"

        normalized_parts: list[str] = []
        for part in raw.parts:
            if part in {"", ".", "/"}:
                continue
            if part == "..":
                if normalized_parts:
                    normalized_parts.pop()
                continue
            normalized_parts.append(part)
        normalized = PurePosixPath("/", *normalized_parts)
        root = coerce_posix_path(self.root)
        root_normalized = PurePosixPath(
            "/",
            *[part for part in root.parts if part not in {"/", ""}],
        )
        if normalized == PurePosixPath("/") or normalized == root_normalized:
            if reject_root:
                raise ValueError("Mount credential exposure path must identify a non-root path.")
            return None
        return f"absolute:{normalized.as_posix()}"

    def ephemeral_entry_paths(self, depth: int | None = 1) -> set[Path]:
        _ = depth
        return {path for path, artifact in self.iter_entries() if artifact.ephemeral}

    def mount_targets(self) -> list[tuple[Mount, Path]]:
        root = posix_path_as_path(coerce_posix_path(self.root))
        mounts: list[tuple[Mount, Path]] = []
        for rel_path, artifact in self.iter_entries():
            if not isinstance(artifact, Mount):
                continue
            dest = resolve_workspace_path(root, rel_path)
            mount_path = artifact._resolve_mount_path_for_root(root, dest)
            normalized_mount_path = self._normalize_in_workspace_path(root, mount_path)
            if normalized_mount_path is not None:
                mount_path = normalized_mount_path
            mounts.append((artifact, mount_path))
        mounts.sort(key=lambda item: len(item[1].parts), reverse=True)
        return mounts

    def ephemeral_mount_targets(self) -> list[tuple[Mount, Path]]:
        return [(artifact, path) for artifact, path in self.mount_targets() if artifact.ephemeral]

    def ephemeral_persistence_paths(self, depth: int | None = 1) -> set[Path]:
        _ = depth
        root = posix_path_as_path(coerce_posix_path(self.root))
        skip = self.ephemeral_entry_paths(depth=depth)
        for _mount, mount_path in self.ephemeral_mount_targets():
            try:
                rel_mount_path = mount_path.relative_to(root)
            except ValueError:
                continue
            if rel_mount_path.parts:
                skip.add(rel_mount_path)
        return skip

    @staticmethod
    def _coerce_rel_path(path: str | PurePath) -> Path:
        if (windows_path := windows_absolute_path(path)) is not None:
            raise InvalidManifestPathError(rel=windows_path.as_posix(), reason="absolute")
        return posix_path_as_path(coerce_posix_path(path))

    @staticmethod
    def _validate_rel_path(rel: Path) -> None:
        if (windows_path := windows_absolute_path(rel)) is not None:
            raise InvalidManifestPathError(rel=windows_path.as_posix(), reason="absolute")
        rel_path = coerce_posix_path(rel)
        if rel_path.is_absolute():
            raise InvalidManifestPathError(rel=rel_path.as_posix(), reason="absolute")
        if ".." in rel_path.parts:
            raise InvalidManifestPathError(rel=rel_path.as_posix(), reason="escape_root")

    @staticmethod
    def _normalize_rel_path_within_root(rel: Path, *, original: Path) -> Path:
        rel_path = coerce_posix_path(rel)
        original_path = coerce_posix_path(original)
        if (windows_path := windows_absolute_path(original)) is not None:
            raise InvalidManifestPathError(rel=windows_path.as_posix(), reason="absolute")
        if rel_path.is_absolute():
            raise InvalidManifestPathError(rel=original_path.as_posix(), reason="absolute")

        normalized_parts: list[str] = []
        for part in rel_path.parts:
            if part in ("", "."):
                continue
            if part == "..":
                if not normalized_parts:
                    raise InvalidManifestPathError(
                        rel=original_path.as_posix(), reason="escape_root"
                    )
                normalized_parts.pop()
                continue
            normalized_parts.append(part)

        return posix_path_as_path(PurePosixPath(*normalized_parts))

    @classmethod
    def _normalize_in_workspace_path(cls, root: Path, path: Path) -> Path | None:
        root_path = coerce_posix_path(root)
        if (windows_path := windows_absolute_path(path)) is not None:
            raise InvalidManifestPathError(rel=windows_path.as_posix(), reason="absolute")
        path_posix = coerce_posix_path(path)
        if not path_posix.is_absolute():
            normalized_rel = cls._normalize_rel_path_within_root(
                posix_path_as_path(path_posix),
                original=posix_path_as_path(path_posix),
            )
            return root / normalized_rel if normalized_rel.parts else root

        try:
            rel_path = path_posix.relative_to(root_path)
        except ValueError:
            return None

        normalized_rel = cls._normalize_rel_path_within_root(
            posix_path_as_path(rel_path),
            original=posix_path_as_path(path_posix),
        )
        root_as_path = posix_path_as_path(root_path)
        return root_as_path / normalized_rel if normalized_rel.parts else root_as_path

    def iter_entries(self) -> Iterator[tuple[Path, BaseEntry]]:
        stack = [
            (self._coerce_rel_path(path), artifact)
            for path, artifact in reversed(list(self.entries.items()))
        ]
        while stack:
            rel_path, artifact = stack.pop()
            self._validate_rel_path(rel_path)
            yield rel_path, artifact
            if not isinstance(artifact, Dir):
                continue

            for child_name, child_artifact in reversed(list(artifact.children.items())):
                child_rel_path = rel_path / self._coerce_rel_path(child_name)
                stack.append((child_rel_path, child_artifact))

    def describe(self, depth: int | None = 1) -> str:
        """
        print a nice fs representation of things inside root with inline descriptions
        depth controls how deep the tree is rendered; None renders all levels
        eg:

        /workspace                      (root)
        ├── repo/                       # /workspace/repo — my repo
        │   └── README.md               # /workspace/repo/README.md
        ├── data/                       # /workspace/data
        │   └── config.json             # /workspace/data/config.json — config
        ├── mount-data/                 # /workspace/mount-data (mount)
        └── notes.txt                   # /workspace/notes.txt
        ...
        """
        return render_manifest_description(
            root=self.root,
            entries=self.validated_entries(),
            coerce_rel_path=self._coerce_rel_path,
            depth=depth,
        )

with_in_container_mount_credential_exposure_acknowledged

with_in_container_mount_credential_exposure_acknowledged(
    *mount_paths: str | PurePath,
) -> Manifest

Acknowledge mount-scoped credential exposure for exact in-container mount paths.

This trusted application-side policy is runtime-only and is not serialized.

Source code in src/agents/sandbox/manifest.py

@redact_mount_validation_error_data_sync
def with_in_container_mount_credential_exposure_acknowledged(
    self, *mount_paths: str | PurePath
) -> "Manifest":
    """Acknowledge mount-scoped credential exposure for exact in-container mount paths.

    This trusted application-side policy is runtime-only and is not serialized.
    """

    return self._with_mount_credential_exposure_acknowledged(
        "mount_scoped",
        mount_paths,
    )

with_in_container_mount_broad_credential_exposure_acknowledged

with_in_container_mount_broad_credential_exposure_acknowledged(
    *mount_paths: str | PurePath,
) -> Manifest

Acknowledge broad credential exposure for exact in-container mount paths.

Broad authority includes managed or workload identity and external credential files. This trusted application-side policy is runtime-only and is not serialized.

Source code in src/agents/sandbox/manifest.py

@redact_mount_validation_error_data_sync
def with_in_container_mount_broad_credential_exposure_acknowledged(
    self, *mount_paths: str | PurePath
) -> "Manifest":
    """Acknowledge broad credential exposure for exact in-container mount paths.

    Broad authority includes managed or workload identity and external credential files.
    This trusted application-side policy is runtime-only and is not serialized.
    """

    return self._with_mount_credential_exposure_acknowledged(
        "broad",
        mount_paths,
    )

describe

describe(depth: int | None = 1) -> str

print a nice fs representation of things inside root with inline descriptions depth controls how deep the tree is rendered; None renders all levels eg:

/workspace (root) ├── repo/ # /workspace/repo — my repo │ └── README.md # /workspace/repo/README.md ├── data/ # /workspace/data │ └── config.json # /workspace/data/config.json — config ├── mount-data/ # /workspace/mount-data (mount) └── notes.txt # /workspace/notes.txt ...

Source code in src/agents/sandbox/manifest.py

def describe(self, depth: int | None = 1) -> str:
    """
    print a nice fs representation of things inside root with inline descriptions
    depth controls how deep the tree is rendered; None renders all levels
    eg:

    /workspace                      (root)
    ├── repo/                       # /workspace/repo — my repo
    │   └── README.md               # /workspace/repo/README.md
    ├── data/                       # /workspace/data
    │   └── config.json             # /workspace/data/config.json — config
    ├── mount-data/                 # /workspace/mount-data (mount)
    └── notes.txt                   # /workspace/notes.txt
    ...
    """
    return render_manifest_description(
        root=self.root,
        entries=self.validated_entries(),
        coerce_rel_path=self._coerce_rel_path,
        depth=depth,
    )

SandboxRunConfig `dataclass`

Grouped sandbox runtime configuration for Runner.

Source code in src/agents/run_config.py

@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."""

    archive_limits: SandboxArchiveLimits | None = None
    """Resource limits for sandbox archive extraction.

    Set to `None` to preserve the default behavior with no SDK archive resource limits.
    Use `SandboxArchiveLimits()` to enable SDK defaults.
    """

    if TYPE_CHECKING:

        def __init__(
            self,
            client: BaseSandboxClient[Any] | None = None,
            options: Any | None = None,
            session: BaseSandboxSession | None = None,
            session_state: SandboxSessionState | None = None,
            manifest: Manifest | dict[str, Any] | None = None,
            snapshot: SnapshotSpec | SnapshotBase | dict[str, Any] | None = None,
            concurrency_limits: SandboxConcurrencyLimits | dict[str, Any] = ...,
            archive_limits: SandboxArchiveLimits | dict[str, Any] | None = None,
        ) -> None: ...

    def __post_init__(self) -> None:
        if isinstance(self.manifest, dict):
            from .sandbox.manifest import _coerce_manifest

            self.manifest = _coerce_manifest(self.manifest, parameter_name="sandbox.manifest")
        if isinstance(self.snapshot, dict):
            from .sandbox.snapshot import SnapshotBase, SnapshotSpecUnion

            if "id" in self.snapshot:
                self.snapshot = SnapshotBase.parse(self.snapshot)
            else:
                self.snapshot = TypeAdapter(SnapshotSpecUnion).validate_python(self.snapshot)
        if isinstance(self.options, dict) and self.client is not None:
            from .sandbox.session.sandbox_client import BaseSandboxClientOptions

            options_type = BaseSandboxClientOptions._options_class_for_type(self.client.backend_id)
            if options_type is not None:
                options = self.options
                explicit_type = options.get("type")
                if explicit_type is not None and explicit_type != self.client.backend_id:
                    raise ValueError(
                        f"sandbox.options type `{explicit_type}` does not match selected "
                        f"sandbox client backend `{self.client.backend_id}`"
                    )
                if "type" not in options:
                    options = {
                        **options,
                        "type": options_type.model_fields["type"].default,
                    }
                self.options = coerce_pydantic_config(
                    options,
                    options_type,
                    parameter_name="sandbox.options",
                )
            elif self.client.backend_id == "blaxel":
                from .extensions.sandbox.blaxel.sandbox import (
                    BlaxelSandboxClient,
                    BlaxelSandboxClientOptions,
                )

                if isinstance(self.client, BlaxelSandboxClient):
                    self.options = coerce_dataclass_config(
                        self.options,
                        BlaxelSandboxClientOptions,
                        parameter_name="sandbox.options",
                    )
        self.concurrency_limits = coerce_dataclass_config(
            self.concurrency_limits,
            _declared_dataclass_type(
                type(self),
                "concurrency_limits",
                SandboxConcurrencyLimits,
            ),
            parameter_name="sandbox.concurrency_limits",
        )
        if self.archive_limits is not None:
            self.archive_limits = coerce_dataclass_config(
                self.archive_limits,
                _declared_dataclass_type(
                    type(self),
                    "archive_limits",
                    SandboxArchiveLimits,
                ),
                parameter_name="sandbox.archive_limits",
            )

client `class-attribute` `instance-attribute`

client: BaseSandboxClient[Any] | None = None

Sandbox client used to create or resume sandbox sessions.

options `class-attribute` `instance-attribute`

options: Any | None = None

Sandbox-client-specific options used when creating a fresh session.

session `class-attribute` `instance-attribute`

session: BaseSandboxSession | None = None

Live sandbox session override for the current process.

session_state `class-attribute` `instance-attribute`

session_state: SandboxSessionState | None = None

Explicit sandbox session state to resume from when not using RunState payloads.

manifest `class-attribute` `instance-attribute`

manifest: Manifest | None = None

Optional sandbox manifest override for fresh session creation.

snapshot `class-attribute` `instance-attribute`

snapshot: SnapshotSpec | SnapshotBase | None = None

Optional sandbox snapshot used for fresh session creation.

concurrency_limits `class-attribute` `instance-attribute`

concurrency_limits: SandboxConcurrencyLimits = field(
    default_factory=SandboxConcurrencyLimits
)

Concurrency limits for sandbox materialization work.

archive_limits `class-attribute` `instance-attribute`

archive_limits: SandboxArchiveLimits | None = None

Resource limits for sandbox archive extraction.

Set to None to preserve the default behavior with no SDK archive resource limits. Use SandboxArchiveLimits() to enable SDK defaults.