Testing

Deterministic test doubles for Agents SDK workflows.

InvalidModelStep

Bases: ModelScriptError

Raised when a model step is invalid before it enters the script queue.

Source code in src/agents/testing/model.py

class InvalidModelStep(ModelScriptError):
    """Raised when a model step is invalid before it enters the script queue."""

    def __init__(
        self,
        message: str,
        *,
        reason: ModelStepReason,
        input_index: int,
    ) -> None:
        super().__init__(message)
        self.reason = reason
        self.input_index = input_index

ModelCall dataclass

A recorded call at the provider-neutral Model boundary.

Source code in src/agents/testing/model.py

@dataclass(frozen=True)
class ModelCall:
    """A recorded call at the provider-neutral ``Model`` boundary."""

    system_instructions: str | None
    input: Any
    model_settings: ModelSettings
    tools: list[Tool]
    output_schema: AgentOutputSchemaBase | None
    handoffs: list[Handoff]
    tracing: ModelTracing
    previous_response_id: str | None
    conversation_id: str | None
    prompt: ResponsePromptParam | None
    streamed: bool

ModelScriptError

Bases: Exception

Base exception for an invalid or incompletely consumed model script.

Source code in src/agents/testing/model.py

class ModelScriptError(Exception):
    """Base exception for an invalid or incompletely consumed model script."""

ModelStep dataclass

One deterministic model call result.

output uses the normalized SDK output-item boundary. Set error to raise from the model call, responder to derive the result from the recorded call, or stream_events to supply an exact normalized event stream for advanced streaming tests. ScriptedModel also accepts the equivalent dictionary form described by ModelStepSpec.

Source code in src/agents/testing/model.py

@dataclass
class ModelStep:
    """One deterministic model call result.

    ``output`` uses the normalized SDK output-item boundary. Set ``error`` to raise from the model
    call, ``responder`` to derive the result from the recorded call, or ``stream_events`` to supply
    an exact normalized event stream for advanced streaming tests. ``ScriptedModel`` also accepts
    the equivalent dictionary form described by ``ModelStepSpec``.
    """

    output: Sequence[TResponseOutputItem] = field(default_factory=tuple)
    usage: Usage = field(default_factory=Usage)
    response_id: str | None = "resp-789"
    request_id: str | None = None
    raw_usage: dict[str, Any] | None = None
    error: Exception | None = None
    responder: ModelResponder | None = None
    stream_events: Sequence[TResponseStreamEvent] | ModelStreamFactory | None = None
    retry_advice: ModelRetryAdvice | None = None

    @classmethod
    def raise_error(
        cls,
        error: Exception,
        *,
        retry_advice: ModelRetryAdvice | None = None,
    ) -> ModelStep:
        """Create a step that raises ``error`` with optional provider retry guidance."""
        return cls(error=error, retry_advice=retry_advice)

    @classmethod
    def respond(cls, responder: ModelResponder) -> ModelStep:
        """Create a step whose result is derived from the recorded call."""
        return cls(responder=responder)

    @classmethod
    def stream(
        cls,
        events: Sequence[TResponseStreamEvent] | ModelStreamFactory,
        *,
        output: Sequence[TResponseOutputItem] = (),
        usage: Usage | None = None,
        response_id: str | None = "resp-789",
    ) -> ModelStep:
        """Create a step with an exact normalized stream-event sequence or factory."""
        stream_events = events if callable(events) else tuple(events)
        return cls(
            output=output,
            usage=usage or Usage(),
            response_id=response_id,
            stream_events=stream_events,
        )

raise_error classmethod

raise_error(
    error: Exception,
    *,
    retry_advice: ModelRetryAdvice | None = None,
) -> ModelStep

Create a step that raises error with optional provider retry guidance.

Source code in src/agents/testing/model.py

@classmethod
def raise_error(
    cls,
    error: Exception,
    *,
    retry_advice: ModelRetryAdvice | None = None,
) -> ModelStep:
    """Create a step that raises ``error`` with optional provider retry guidance."""
    return cls(error=error, retry_advice=retry_advice)

respond classmethod

respond(responder: ModelResponder) -> ModelStep

Create a step whose result is derived from the recorded call.

Source code in src/agents/testing/model.py

@classmethod
def respond(cls, responder: ModelResponder) -> ModelStep:
    """Create a step whose result is derived from the recorded call."""
    return cls(responder=responder)

stream classmethod

stream(
    events: Sequence[TResponseStreamEvent]
    | ModelStreamFactory,
    *,
    output: Sequence[TResponseOutputItem] = (),
    usage: Usage | None = None,
    response_id: str | None = "resp-789",
) -> ModelStep

Create a step with an exact normalized stream-event sequence or factory.

Source code in src/agents/testing/model.py

@classmethod
def stream(
    cls,
    events: Sequence[TResponseStreamEvent] | ModelStreamFactory,
    *,
    output: Sequence[TResponseOutputItem] = (),
    usage: Usage | None = None,
    response_id: str | None = "resp-789",
) -> ModelStep:
    """Create a step with an exact normalized stream-event sequence or factory."""
    stream_events = events if callable(events) else tuple(events)
    return cls(
        output=output,
        usage=usage or Usage(),
        response_id=response_id,
        stream_events=stream_events,
    )

ModelStepSpec

Bases: TypedDict

Dictionary form of ModelStep accepted by ScriptedModel.

Source code in src/agents/testing/model.py

class ModelStepSpec(TypedDict, total=False):
    """Dictionary form of ``ModelStep`` accepted by ``ScriptedModel``."""

    output: Sequence[TResponseOutputItem]
    usage: Usage
    response_id: str | None
    request_id: str | None
    raw_usage: dict[str, Any] | None
    error: Exception | None
    responder: ModelResponder | None
    stream_events: Sequence[TResponseStreamEvent] | ModelStreamFactory | None
    retry_advice: ModelRetryAdvice | None

ScriptedModel

Bases: Model

A deterministic provider-neutral model for testing agent workflows.

Each step may be a ModelStep, an equivalent ModelStepSpec dictionary, a ModelResponse, a normalized output-item sequence, or an exception.

Source code in src/agents/testing/model.py

class ScriptedModel(Model):
    """A deterministic provider-neutral model for testing agent workflows.

    Each step may be a ``ModelStep``, an equivalent ``ModelStepSpec`` dictionary, a
    ``ModelResponse``, a normalized output-item sequence, or an exception.
    """

    def __init__(
        self,
        steps: Iterable[ModelScriptItem] = (),
        *,
        emit_traces: bool = False,
        default_usage: Usage | None = None,
    ) -> None:
        self._steps = [
            self._coerce_step(step, input_index) for input_index, step in enumerate(steps)
        ]
        self._emit_traces = emit_traces
        self._default_usage = copy.deepcopy(default_usage)
        self._calls: list[ModelCall] = []
        self._retry_advice_by_error_id: dict[int, tuple[Exception, ModelRetryAdvice]] = {}

    @property
    def calls(self) -> tuple[ModelCall, ...]:
        """Return detached snapshots of recorded model calls."""
        return tuple(_snapshot_model_call(call) for call in self._calls)

    @property
    def remaining_steps(self) -> int:
        """Return the number of configured model calls that have not run yet."""
        return len(self._steps)

    @property
    def first_call(self) -> ModelCall | None:
        """Return the first recorded call, if any."""
        return _snapshot_model_call(self._calls[0]) if self._calls else None

    @property
    def last_call(self) -> ModelCall | None:
        """Return the most recent recorded call, if any."""
        return _snapshot_model_call(self._calls[-1]) if self._calls else None

    def enqueue(self, step: ModelScriptItem) -> None:
        """Append one model step."""
        self._steps.append(self._coerce_step(step, 0))

    def extend(self, steps: Iterable[ModelScriptItem]) -> None:
        """Append multiple model steps."""
        normalized = [
            self._coerce_step(step, input_index) for input_index, step in enumerate(steps)
        ]
        self._steps.extend(normalized)

    def set_default_usage(self, usage: Usage | None) -> None:
        """Set usage for scripted steps that do not provide their own usage."""
        self._default_usage = copy.deepcopy(usage)

    def get_retry_advice(self, request: ModelRetryAdviceRequest) -> ModelRetryAdvice | None:
        """Return retry advice attached to the exact scripted error that was raised."""
        configured = self._retry_advice_by_error_id.get(id(request.error))
        if configured is None or configured[0] is not request.error:
            return None
        return copy.deepcopy(configured[1])

    def assert_complete(self) -> None:
        """Raise when configured steps remain unconsumed."""
        if self._steps:
            raise UnconsumedModelSteps(
                f"{len(self._steps)} scripted model step(s) were not consumed.",
                remaining_steps=len(self._steps),
            )

    async def get_response(
        self,
        system_instructions: str | None,
        input: str | list[TResponseInputItem],
        model_settings: ModelSettings,
        tools: list[Tool],
        output_schema: AgentOutputSchemaBase | None,
        handoffs: list[Handoff],
        tracing: ModelTracing,
        *,
        previous_response_id: str | None,
        conversation_id: str | None,
        prompt: ResponsePromptParam | None,
    ) -> ModelResponse:
        call = self._record_call(
            system_instructions=system_instructions,
            input=input,
            model_settings=model_settings,
            tools=tools,
            output_schema=output_schema,
            handoffs=handoffs,
            tracing=tracing,
            previous_response_id=previous_response_id,
            conversation_id=conversation_id,
            prompt=prompt,
            streamed=False,
        )
        with generation_span(disabled=not self._emit_traces) as span:
            retry_advice_synced = False
            try:
                step = await self._next_resolved_step(call)
                if step.error is not None:
                    self._remember_retry_advice(step)
                    retry_advice_synced = True
                    raise step.error
                return self._model_response(step, call.model_settings)
            except Exception as error:
                if not retry_advice_synced:
                    self._forget_retry_advice(error)
                self._set_span_error(span, error, call.tracing)
                raise

    async def stream_response(
        self,
        system_instructions: str | None,
        input: str | list[TResponseInputItem],
        model_settings: ModelSettings,
        tools: list[Tool],
        output_schema: AgentOutputSchemaBase | None,
        handoffs: list[Handoff],
        tracing: ModelTracing,
        *,
        previous_response_id: str | None,
        conversation_id: str | None,
        prompt: ResponsePromptParam | None,
    ) -> AsyncIterator[TResponseStreamEvent]:
        call = self._record_call(
            system_instructions=system_instructions,
            input=input,
            model_settings=model_settings,
            tools=tools,
            output_schema=output_schema,
            handoffs=handoffs,
            tracing=tracing,
            previous_response_id=previous_response_id,
            conversation_id=conversation_id,
            prompt=prompt,
            streamed=True,
        )
        span = generation_span(disabled=not self._emit_traces)
        span.start(mark_as_current=False)
        retry_advice_synced = False
        try:
            with _mark_span_current(span):
                step = await self._next_resolved_step(call)
            if step.error is not None:
                self._remember_retry_advice(step)
                retry_advice_synced = True
                raise step.error
            if callable(step.stream_events):
                with _mark_span_current(span):
                    stream = step.stream_events(call)
                try:
                    while True:
                        try:
                            with _mark_span_current(span):
                                event = await anext(stream)
                        except StopAsyncIteration:
                            break
                        yield event
                finally:
                    aclose = getattr(stream, "aclose", None)
                    if callable(aclose):
                        active_error = sys.exc_info()[1]
                        try:
                            with _mark_span_current(span):
                                await aclose()
                        except BaseException:
                            if active_error is None:
                                raise
                return
            if step.stream_events is not None:
                for event in step.stream_events:
                    yield event
                return
            with _mark_span_current(span):
                events = _stream_events_for_step(
                    step,
                    preserve_raw_usage=call.model_settings.preserve_raw_usage is True,
                )
            for event in events:
                yield event
        except Exception as error:
            if not retry_advice_synced:
                self._forget_retry_advice(error)
            self._set_span_error(span, error, call.tracing)
            raise
        finally:
            span.finish(reset_current=False)

    def _record_call(
        self,
        *,
        system_instructions: str | None,
        input: str | list[TResponseInputItem],
        model_settings: ModelSettings,
        tools: list[Tool],
        output_schema: AgentOutputSchemaBase | None,
        handoffs: list[Handoff],
        tracing: ModelTracing,
        previous_response_id: str | None,
        conversation_id: str | None,
        prompt: ResponsePromptParam | None,
        streamed: bool,
    ) -> ModelCall:
        recorded_input = copy.deepcopy(input)
        call = ModelCall(
            system_instructions=system_instructions,
            input=recorded_input,
            model_settings=copy.deepcopy(model_settings),
            tools=list(tools),
            output_schema=output_schema,
            handoffs=list(handoffs),
            tracing=tracing,
            previous_response_id=previous_response_id,
            conversation_id=conversation_id,
            prompt=copy.deepcopy(prompt),
            streamed=streamed,
        )
        execution_call = _snapshot_model_call(call)
        self._calls.append(call)
        return execution_call

    async def _next_resolved_step(self, call: ModelCall) -> ModelStep:
        if not self._steps:
            mode = "streaming" if call.streamed else "non-streaming"
            call_index = len(self._calls) - 1
            raise UnexpectedModelCall(
                f"Unexpected {mode} model call #{call_index + 1}: no scripted steps remain.",
                call=call,
                call_index=call_index,
            )
        step = self._steps.pop(0)
        while step.responder is not None:
            result = step.responder(call)
            if inspect.isawaitable(result):
                result = await result
            step = self._coerce_step(result, 0)
        if step.usage == Usage():
            usage = self._default_usage if self._default_usage is not None else Usage(requests=1)
        else:
            usage = step.usage
        return replace(step, usage=copy.deepcopy(usage))

    @staticmethod
    def _coerce_step(
        step: ModelScriptItem | ModelStepResult,
        input_index: int,
    ) -> ModelStep:
        if isinstance(step, ModelStep):
            normalized = step
        elif isinstance(step, ModelResponse):
            normalized = ModelStep(
                output=step.output,
                usage=step.usage,
                response_id=step.response_id,
                request_id=step.request_id,
                raw_usage=step.raw_usage,
            )
        elif isinstance(step, Exception):
            normalized = ModelStep.raise_error(step)
        elif isinstance(step, Mapping):
            unsupported = [field for field in step if field not in _MODEL_STEP_FIELDS]
            if unsupported:
                raise _invalid_model_step(
                    reason="unsupported_field",
                    input_index=input_index,
                    detail="contains unsupported fields",
                )
            normalized = ModelStep(**cast(ModelStepSpec, dict(step)))
        else:
            normalized = ModelStep(output=step)
        _validate_model_step(normalized, input_index=input_index)
        return _snapshot_model_step(normalized)

    def _remember_retry_advice(self, step: ModelStep) -> None:
        if step.error is None:
            return
        if step.retry_advice is None:
            self._forget_retry_advice(step.error)
            return
        self._retry_advice_by_error_id[id(step.error)] = (
            step.error,
            copy.deepcopy(step.retry_advice),
        )

    def _forget_retry_advice(self, error: Exception) -> None:
        self._retry_advice_by_error_id.pop(id(error), None)

    @staticmethod
    def _model_response(step: ModelStep, model_settings: ModelSettings) -> ModelResponse:
        return ModelResponse(
            output=_convert_output_items(step.output),
            usage=step.usage,
            response_id=step.response_id,
            request_id=step.request_id,
            raw_usage=(
                _raw_usage_snapshot(step.raw_usage)
                if model_settings.preserve_raw_usage is True
                else None
            ),
        )

    @staticmethod
    def _set_span_error(span: Any, error: Exception, tracing: ModelTracing) -> None:
        try:
            if tracing.include_data():
                try:
                    error_message = str(error)
                except BaseException:
                    error_message = f"Unrenderable {type(error).__name__}"
            else:
                error_message = REDACTED_TRACE_ERROR_MESSAGE
            span.set_error(
                SpanError(
                    message="Error",
                    data={"name": error.__class__.__name__, "message": error_message},
                )
            )
        except BaseException:
            pass

calls property

calls: tuple[ModelCall, ...]

Return detached snapshots of recorded model calls.

remaining_steps property

remaining_steps: int

Return the number of configured model calls that have not run yet.

first_call property

first_call: ModelCall | None

Return the first recorded call, if any.

last_call property

last_call: ModelCall | None

Return the most recent recorded call, if any.

close async

close() -> None

Release any resources held by the model.

Models that maintain persistent connections can override this. The default implementation is a no-op.

Source code in src/agents/models/interface.py

async def close(self) -> None:
    """Release any resources held by the model.

    Models that maintain persistent connections can override this. The default implementation
    is a no-op.
    """
    return None

enqueue

enqueue(step: ModelScriptItem) -> None

Append one model step.

Source code in src/agents/testing/model.py

def enqueue(self, step: ModelScriptItem) -> None:
    """Append one model step."""
    self._steps.append(self._coerce_step(step, 0))

extend

extend(steps: Iterable[ModelScriptItem]) -> None

Append multiple model steps.

Source code in src/agents/testing/model.py

def extend(self, steps: Iterable[ModelScriptItem]) -> None:
    """Append multiple model steps."""
    normalized = [
        self._coerce_step(step, input_index) for input_index, step in enumerate(steps)
    ]
    self._steps.extend(normalized)

set_default_usage

set_default_usage(usage: Usage | None) -> None

Set usage for scripted steps that do not provide their own usage.

Source code in src/agents/testing/model.py

def set_default_usage(self, usage: Usage | None) -> None:
    """Set usage for scripted steps that do not provide their own usage."""
    self._default_usage = copy.deepcopy(usage)

get_retry_advice

get_retry_advice(
    request: ModelRetryAdviceRequest,
) -> ModelRetryAdvice | None

Return retry advice attached to the exact scripted error that was raised.

Source code in src/agents/testing/model.py

def get_retry_advice(self, request: ModelRetryAdviceRequest) -> ModelRetryAdvice | None:
    """Return retry advice attached to the exact scripted error that was raised."""
    configured = self._retry_advice_by_error_id.get(id(request.error))
    if configured is None or configured[0] is not request.error:
        return None
    return copy.deepcopy(configured[1])

assert_complete

assert_complete() -> None

Raise when configured steps remain unconsumed.

Source code in src/agents/testing/model.py

def assert_complete(self) -> None:
    """Raise when configured steps remain unconsumed."""
    if self._steps:
        raise UnconsumedModelSteps(
            f"{len(self._steps)} scripted model step(s) were not consumed.",
            remaining_steps=len(self._steps),
        )

UnconsumedModelSteps

Bases: ModelScriptError

Raised when a test finishes before consuming every configured step.

Source code in src/agents/testing/model.py

class UnconsumedModelSteps(ModelScriptError):
    """Raised when a test finishes before consuming every configured step."""

    def __init__(self, message: str, *, remaining_steps: int) -> None:
        super().__init__(message)
        self.remaining_steps = remaining_steps

UnexpectedModelCall

Bases: ModelScriptError

Raised when the model is called after all configured steps were consumed.

Source code in src/agents/testing/model.py

class UnexpectedModelCall(ModelScriptError):
    """Raised when the model is called after all configured steps were consumed."""

    def __init__(self, message: str, *, call: ModelCall, call_index: int) -> None:
        super().__init__(message)
        self.call = call
        self.call_index = call_index

InvalidSandboxStep

Bases: SandboxScriptError

Raised when a sandbox step is invalid at factory construction time.

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

class InvalidSandboxStep(SandboxScriptError):
    """Raised when a sandbox step is invalid at factory construction time."""

    def __init__(
        self,
        message: str,
        *,
        reason: SandboxStepReason,
        input_index: int,
        method: str | None,
    ) -> None:
        super().__init__(message)
        self.reason = reason
        self.input_index = input_index
        self.method = method

SandboxCall dataclass

A detached invocation-time sandbox call snapshot.

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

@dataclass(frozen=True)
class SandboxCall:
    """A detached invocation-time sandbox call snapshot."""

    call_index: int
    method: str
    args: tuple[Any, ...]
    kwargs: Mapping[str, Any]

SandboxCallMatcherError

Bases: SandboxScriptError

Raised when a sandbox step matcher rejects its call.

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

class SandboxCallMatcherError(SandboxScriptError):
    """Raised when a sandbox step matcher rejects its call."""

    def __init__(self, message: str, *, call: SandboxCall, call_index: int) -> None:
        super().__init__(message)
        self.call = call
        self.call_index = call_index
        self.method: str = call.method

SandboxScriptError

Bases: Exception

Base exception for an invalid or incompletely consumed sandbox script.

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

class SandboxScriptError(Exception):
    """Base exception for an invalid or incompletely consumed sandbox script."""

SandboxStepSpec

Bases: TypedDict

Dictionary form of one FIFO scripted sandbox call.

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

class SandboxStepSpec(TypedDict, total=False):
    """Dictionary form of one FIFO scripted sandbox call."""

    method: SandboxMethod
    match: SandboxMatcher
    result: Any
    responder: SandboxResponder
    error: Exception

ScriptedSandboxSession

Bases: BaseSandboxSession, ABC

The typed result interface for scripted_sandbox_session.

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

class ScriptedSandboxSession(BaseSandboxSession, abc.ABC):
    """The typed result interface for ``scripted_sandbox_session``."""

    @property
    @abc.abstractmethod
    def calls(self) -> tuple[SandboxCall, ...]:
        """Return detached call-history snapshots in invocation order."""

    @property
    @abc.abstractmethod
    def remaining_steps(self) -> int:
        """Return the number of configured calls that remain."""

    @abc.abstractmethod
    def assert_complete(self) -> None:
        """Raise when configured sandbox calls remain unconsumed."""

calls abstractmethod property

calls: tuple[SandboxCall, ...]

Return detached call-history snapshots in invocation order.

remaining_steps abstractmethod property

remaining_steps: int

Return the number of configured calls that remain.

stop async

stop() -> None

Persist/snapshot the workspace.

Note: stop() is intentionally persistence-only. Sandboxes that need to tear down sandbox resources (Docker containers, remote sessions, etc.) should implement shutdown() instead.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@redact_mount_error_data
async def stop(self) -> None:
    """
    Persist/snapshot the workspace.

    Note: `stop()` is intentionally persistence-only. Sandboxes that need to tear down
    sandbox resources (Docker containers, remote sessions, etc.) should implement
    `shutdown()` instead.
    """
    validate_manifest_mount_credential_boundaries(
        self.state.manifest,
        provider_backend_id=self.state.type,
    )
    try:
        try:
            await self._before_stop()
            await self._persist_snapshot()
        except Exception as e:
            wrapped = self._wrap_stop_error(e)
            if wrapped is e:
                raise
            raise wrapped from e
    finally:
        await self._after_stop()

supports_docker_volume_mounts

supports_docker_volume_mounts() -> bool

Return whether this backend attaches Docker volume mounts before manifest apply.

Source code in src/agents/sandbox/session/base_sandbox_session.py

def supports_docker_volume_mounts(self) -> bool:
    """Return whether this backend attaches Docker volume mounts before manifest apply."""

    return False

shutdown async

shutdown() -> None

Tear down sandbox resources (best-effort).

Default is a no-op. Sandbox-specific sessions (e.g. Docker) should override.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@redact_mount_error_data
async def shutdown(self) -> None:
    """
    Tear down sandbox resources (best-effort).

    Default is a no-op. Sandbox-specific sessions (e.g. Docker) should override.
    """
    await self._before_shutdown()
    await self._shutdown_backend()
    await self._after_shutdown()

aclose async

aclose() -> None

Run the session cleanup lifecycle outside of async with.

This performs the same session-owned cleanup as __aexit__(): persist/snapshot the workspace via stop(), tear down session resources via shutdown(), and close session-scoped dependencies. If the session came from a sandbox client, call the client's delete() separately for backend-specific deletion such as removing a Docker container or deleting a temporary host workspace.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@redact_mount_error_data
async def aclose(self) -> None:
    """Run the session cleanup lifecycle outside of ``async with``.

    This performs the same session-owned cleanup as ``__aexit__()``: persist/snapshot the
    workspace via ``stop()``, tear down session resources via ``shutdown()``, and close
    session-scoped dependencies. If the session came from a sandbox client, call the client's
    ``delete()`` separately for backend-specific deletion such as removing a Docker container
    or deleting a temporary host workspace.
    """

    lock = self._aclose_lock
    if lock is None:
        lock = asyncio.Lock()
        self._aclose_lock = lock
    async with lock:
        await self._aclose_impl()

register_pre_stop_hook

register_pre_stop_hook(
    hook: Callable[[], Awaitable[None]],
) -> None

Register an async hook to run once before the session workspace is persisted.

Source code in src/agents/sandbox/session/base_sandbox_session.py

def register_pre_stop_hook(self, hook: Callable[[], Awaitable[None]]) -> None:
    """Register an async hook to run once before the session workspace is persisted."""

    hooks = self._pre_stop_hooks
    if hooks is None:
        hooks = []
        self._pre_stop_hooks = hooks
    hooks.append(hook)
    self._pre_stop_hooks_ran = False

run_pre_stop_hooks async

run_pre_stop_hooks() -> None

Run registered pre-stop hooks once before workspace persistence.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@redact_mount_error_data
async def run_pre_stop_hooks(self) -> None:
    """Run registered pre-stop hooks once before workspace persistence."""

    lock = self._pre_stop_hooks_lock
    if lock is None:
        lock = asyncio.Lock()
        self._pre_stop_hooks_lock = lock
    async with lock:
        hooks = self._pre_stop_hooks
        if hooks is None or self._pre_stop_hooks_ran:
            return
        self._pre_stop_hooks_ran = True
        cleanup_error: BaseException | None = None
        for hook in hooks:
            try:
                await hook()
            except BaseException as exc:
                if cleanup_error is None:
                    cleanup_error = exc
        if cleanup_error is not None:
            self._pre_stop_hooks_failed = True
            raise cleanup_error

register_persist_workspace_skip_path

register_persist_workspace_skip_path(
    path: Path | str,
) -> Path

Exclude a runtime-created workspace path from future workspace snapshots.

Use this for session side effects that are not part of durable workspace state, such as generated mount config or ephemeral sink output.

Source code in src/agents/sandbox/session/base_sandbox_session.py

def register_persist_workspace_skip_path(self, path: Path | str) -> Path:
    """Exclude a runtime-created workspace path from future workspace snapshots.

    Use this for session side effects that are not part of durable workspace state, such as
    generated mount config or ephemeral sink output.
    """

    rel_path = Manifest._coerce_rel_path(path)
    Manifest._validate_rel_path(rel_path)
    if rel_path in (Path(""), Path(".")):
        raise ValueError("Persist workspace skip paths must target a concrete relative path.")
    overlapping_mounts = self._overlapping_mount_relpaths(rel_path)
    if overlapping_mounts:
        overlapping_mount = min(overlapping_mounts, key=lambda p: (len(p.parts), p.as_posix()))
        raise MountConfigError(
            message="persist workspace skip path must not overlap mount path",
            context={
                "skip_path": rel_path.as_posix(),
                "mount_path": overlapping_mount.as_posix(),
            },
        )

    if self._runtime_persist_workspace_skip_relpaths is None:
        self._runtime_persist_workspace_skip_relpaths = set()
    self._runtime_persist_workspace_skip_relpaths.add(rel_path)
    return rel_path

exec async

exec(
    *command: str | Path,
    timeout: float | None = None,
    shell: bool | list[str] = True,
    user: str | User | None = None,
) -> ExecResult

Execute a command inside the session.

:param command: Command and args (will be stringified). :param timeout: Optional wall-clock timeout in seconds. :param shell: Whether to run this command in a shell. If True is provided, the command will be run prefixed by sh -lc. A custom shell prefix may be used by providing a list.

:returns: An ExecResult containing stdout/stderr and exit code.

:raises TimeoutError: If the sandbox cannot complete within timeout.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@redact_mount_error_data
async def exec(
    self,
    *command: str | Path,
    timeout: float | None = None,
    shell: bool | list[str] = True,
    user: str | User | None = None,
) -> ExecResult:
    """Execute a command inside the session.

    :param command: Command and args (will be stringified).
    :param timeout: Optional wall-clock timeout in seconds.
    :param shell: Whether to run this command in a shell. If ``True`` is provided,
        the command will be run prefixed by ``sh -lc``. A custom shell prefix may be used
        by providing a list.

    :returns: An ``ExecResult`` containing stdout/stderr and exit code.

    :raises TimeoutError: If the sandbox cannot complete within `timeout`.
    """

    sanitized_command = self._prepare_exec_command(*command, shell=shell, user=user)
    return await self._exec_internal(*sanitized_command, timeout=timeout)

read abstractmethod async

read(
    path: Path, *, user: str | User | None = None
) -> IOBase

Read a file from the session's workspace.

:param path: Absolute path in the container or path relative to the workspace root. :param user: Optional sandbox user to perform the read as. :returns: A readable file-like object. :raises: FileNotFoundError: If the path does not exist.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@abc.abstractmethod
async def read(self, path: Path, *, user: str | User | None = None) -> io.IOBase:
    """Read a file from the session's workspace.

    :param path: Absolute path in the container or path relative to the
            workspace root.
    :param user: Optional sandbox user to perform the read as.
    :returns: A readable file-like object.
    :raises: FileNotFoundError: If the path does not exist.
    """

write abstractmethod async

write(
    path: Path,
    data: IOBase,
    *,
    user: str | User | None = None,
) -> None

Write a file into the session's workspace.

:param path: Absolute path in the container or path relative to the workspace root. :param data: A file-like object positioned at the start of the payload. :param user: Optional sandbox user to perform the write as.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@abc.abstractmethod
async def write(
    self,
    path: Path,
    data: io.IOBase,
    *,
    user: str | User | None = None,
) -> None:
    """Write a file into the session's workspace.

    :param path: Absolute path in the container or path relative to the
            workspace root.
    :param data: A file-like object positioned at the start of the payload.
    :param user: Optional sandbox user to perform the write as.
    """

running abstractmethod async

running() -> bool

:returns: whether the underlying sandbox is currently running.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@abc.abstractmethod
async def running(self) -> bool:
    """
    :returns: whether the underlying sandbox is currently running.
    """

persist_workspace abstractmethod async

persist_workspace() -> IOBase

Serialize the session's workspace into a byte stream.

:returns: A readable byte stream representing the workspace contents. Portable tar streams must use workspace-relative member paths rather than embedding the source backend's workspace root directory.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@abc.abstractmethod
async def persist_workspace(self) -> io.IOBase:
    """Serialize the session's workspace into a byte stream.

    :returns: A readable byte stream representing the workspace contents.
        Portable tar streams must use workspace-relative member paths rather than
        embedding the source backend's workspace root directory.
    """

hydrate_workspace abstractmethod async

hydrate_workspace(data: IOBase) -> None

Populate the session's workspace from a serialized byte stream.

:param data: A readable byte stream as produced by persist_workspace. Portable tar streams are extracted underneath this session's workspace root.

Source code in src/agents/sandbox/session/base_sandbox_session.py

@abc.abstractmethod
async def hydrate_workspace(self, data: io.IOBase) -> None:
    """Populate the session's workspace from a serialized byte stream.

    :param data: A readable byte stream as produced by `persist_workspace`.
        Portable tar streams are extracted underneath this session's workspace root.
    """

ls async

ls(
    path: Path | str, *, user: str | User | None = None
) -> list[FileEntry]

List directory contents.

:param path: Path to list. :param user: Optional sandbox user to list as. :returns: A list of FileEntry objects.

Source code in src/agents/sandbox/session/base_sandbox_session.py

async def ls(
    self,
    path: Path | str,
    *,
    user: str | User | None = None,
) -> list[FileEntry]:
    """List directory contents.

    :param path: Path to list.
    :param user: Optional sandbox user to list as.
    :returns: A list of `FileEntry` objects.
    """
    path = await self._validate_path_access(path)

    path_arg = sandbox_path_str(path)
    cmd = ("ls", "-la", "--", path_arg)
    result = await self.exec(*cmd, shell=False, user=user)
    if not result.ok():
        raise ExecNonZeroError(result, command=cmd)

    return parse_ls_la(result.stdout.decode("utf-8", errors="replace"), base=path_arg)

rm async

rm(
    path: Path | str,
    *,
    recursive: bool = False,
    user: str | User | None = None,
) -> None

Remove a file or directory.

:param path: Path to remove. :param recursive: If true, remove directories recursively. :param user: Optional sandbox user to remove as.

Source code in src/agents/sandbox/session/base_sandbox_session.py

async def rm(
    self,
    path: Path | str,
    *,
    recursive: bool = False,
    user: str | User | None = None,
) -> None:
    """Remove a file or directory.

    :param path: Path to remove.
    :param recursive: If true, remove directories recursively.
    :param user: Optional sandbox user to remove as.
    """
    path = await self._validate_path_access(path, for_write=True)

    cmd: list[str] = ["rm"]
    if recursive:
        cmd.append("-rf")
    cmd.extend(["--", sandbox_path_str(path)])

    result = await self.exec(*cmd, shell=False, user=user)
    if not result.ok():
        raise ExecNonZeroError(result, command=cmd)

mkdir async

mkdir(
    path: Path | str,
    *,
    parents: bool = False,
    user: str | User | None = None,
) -> None

Create a directory.

:param path: Directory to create on the remote. :param parents: If true, create missing parents. :param user: Optional sandbox user to create the directory as.

Source code in src/agents/sandbox/session/base_sandbox_session.py

async def mkdir(
    self,
    path: Path | str,
    *,
    parents: bool = False,
    user: str | User | None = None,
) -> None:
    """Create a directory.

    :param path: Directory to create on the remote.
    :param parents: If true, create missing parents.
    :param user: Optional sandbox user to create the directory as.
    """
    path = await self._validate_path_access(path, for_write=True)

    cmd: list[str] = ["mkdir"]
    if parents:
        cmd.append("-p")
    cmd.append(sandbox_path_str(path))

    result = await self.exec(*cmd, shell=False, user=user)
    if not result.ok():
        raise ExecNonZeroError(result, command=cmd)

extract async

extract(
    path: Path | str,
    data: IOBase,
    *,
    compression_scheme: Literal["tar", "zip"] | None = None,
    archive_limits: SandboxArchiveLimits | None = None,
) -> None

Write a compressed archive to a destination on the remote. Optionally extract the archive once written.

:param path: Path on the host machine to extract to :param data: a file-like io stream. :param compression_scheme: either "tar" or "zip". If not provided, it will try to infer from the path. :param archive_limits: optional per-call archive resource limits. If omitted, the session default is used.

Source code in src/agents/sandbox/session/base_sandbox_session.py

async def extract(
    self,
    path: Path | str,
    data: io.IOBase,
    *,
    compression_scheme: Literal["tar", "zip"] | None = None,
    archive_limits: SandboxArchiveLimits | None = None,
) -> None:
    """
    Write a compressed archive to a destination on the remote.
    Optionally extract the archive once written.

    :param path: Path on the host machine to extract to
    :param data: a file-like io stream.
    :param compression_scheme: either "tar" or "zip". If not provided,
        it will try to infer from the path.
    :param archive_limits: optional per-call archive resource limits. If omitted,
        the session default is used.
    """
    if archive_limits is not None:
        archive_limits.validate()
    effective_archive_limits = (
        archive_limits if archive_limits is not None else self._archive_limits
    )

    await archive_ops.extract_archive(
        self,
        path,
        data,
        compression_scheme=compression_scheme,
        archive_limits=effective_archive_limits,
    )

should_provision_manifest_accounts_on_resume

should_provision_manifest_accounts_on_resume() -> bool

Return whether resume should reprovision manifest-managed users and groups.

Source code in src/agents/sandbox/session/base_sandbox_session.py

def should_provision_manifest_accounts_on_resume(self) -> bool:
    """Return whether resume should reprovision manifest-managed users and groups."""

    return not self._system_state_preserved_on_start()

assert_complete abstractmethod

assert_complete() -> None

Raise when configured sandbox calls remain unconsumed.

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

@abc.abstractmethod
def assert_complete(self) -> None:
    """Raise when configured sandbox calls remain unconsumed."""

UnconsumedSandboxSteps

Bases: SandboxScriptError

Raised when configured sandbox steps remain unconsumed.

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

class UnconsumedSandboxSteps(SandboxScriptError):
    """Raised when configured sandbox steps remain unconsumed."""

    def __init__(
        self,
        message: str,
        *,
        remaining_steps: int,
        pending_methods: tuple[str, ...],
    ) -> None:
        super().__init__(message)
        self.remaining_steps = remaining_steps
        self.pending_methods = pending_methods

UnexpectedSandboxCall

Bases: SandboxScriptError

Raised when a call does not match the next configured sandbox step.

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

class UnexpectedSandboxCall(SandboxScriptError):
    """Raised when a call does not match the next configured sandbox step."""

    def __init__(
        self,
        message: str,
        *,
        call: SandboxCall,
        call_index: int,
        expected_method: str | None,
        remaining_steps: int,
    ) -> None:
        super().__init__(message)
        self.call = call
        self.call_index = call_index
        self.actual_method: str = call.method
        self.expected_method = expected_method
        self.remaining_steps = remaining_steps

assistant_message

assistant_message(
    text: str, *, item_id: str = "scripted-message"
) -> TResponseOutputItem

Build one normalized assistant text output item.

Source code in src/agents/testing/model.py

def assistant_message(text: str, *, item_id: str = "scripted-message") -> TResponseOutputItem:
    """Build one normalized assistant text output item."""
    return ResponseOutputMessage(
        id=item_id,
        type="message",
        role="assistant",
        status="completed",
        content=[
            ResponseOutputText(
                text=text,
                type="output_text",
                annotations=[],
                logprobs=[],
            )
        ],
    )

function_call

function_call(
    name: str,
    arguments: str | Mapping[str, Any],
    *,
    call_id: str,
    item_id: str | None = None,
    namespace: str | None = None,
) -> TResponseOutputItem

Build one normalized function-tool call output item.

Source code in src/agents/testing/model.py

def function_call(
    name: str,
    arguments: str | Mapping[str, Any],
    *,
    call_id: str,
    item_id: str | None = None,
    namespace: str | None = None,
) -> TResponseOutputItem:
    """Build one normalized function-tool call output item."""
    serialized_arguments = (
        arguments
        if isinstance(arguments, str)
        else json.dumps(arguments, ensure_ascii=False, separators=(",", ":"))
    )
    kwargs: dict[str, Any] = {
        "id": call_id if item_id is None else item_id,
        "call_id": call_id,
        "type": "function_call",
        "name": name,
        "arguments": serialized_arguments,
    }
    if namespace is not None:
        kwargs["namespace"] = namespace
    return ResponseFunctionToolCall(**kwargs)

scripted_sandbox_session

scripted_sandbox_session(
    steps: Iterable[
        SandboxStepSpec | Mapping[str, Any]
    ] = (),
    *,
    manifest: Manifest | None = None,
) -> ScriptedSandboxSession

Create a deterministic provider-free sandbox session for agent workflow tests.

Each FIFO step defines method plus exactly one of result, responder, or error. An optional match callable receives a detached SandboxCall. The returned object is the session itself, so pass it directly to SandboxRunConfig(session=session). Only configured model-facing methods are visible. The two PTY methods are exposed together when either one is configured because they form one advertised session capability. Use a custom BaseSandboxSession or a real provider for lifecycle, persistence, mount, or broader filesystem behavior.

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

def scripted_sandbox_session(
    steps: Iterable[SandboxStepSpec | Mapping[str, Any]] = (),
    *,
    manifest: Manifest | None = None,
) -> ScriptedSandboxSession:
    """Create a deterministic provider-free sandbox session for agent workflow tests.

    Each FIFO step defines ``method`` plus exactly one of ``result``, ``responder``, or ``error``.
    An optional ``match`` callable receives a detached ``SandboxCall``. The returned object is the
    session itself, so pass it directly to ``SandboxRunConfig(session=session)``. Only configured
    model-facing methods are visible. The two PTY methods are exposed together when either one is
    configured because they form one advertised session capability. Use a custom
    ``BaseSandboxSession`` or a real provider for lifecycle, persistence, mount, or broader
    filesystem behavior.
    """
    normalized = [_normalize_step(step, index) for index, step in enumerate(steps)]
    return _ScriptedSandboxSession(normalized, manifest=manifest)