Model

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

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

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

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

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

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 asyncio.CancelledError:
                record_current_task_model_timeout_on_span(
                    span,
                    message="Error",
                    trace_include_sensitive_data=call.tracing.include_data(),
                )
                raise
            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 asyncio.CancelledError:
            record_current_task_model_timeout_on_span(
                span,
                message="Error",
                trace_include_sensitive_data=call.tracing.include_data(),
            )
            raise
        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.

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),
        )

get_default_model_settings

get_default_model_settings() -> ModelSettings | None

Return defaults for this model, or None to retain the agent's implicit defaults.

The default implementation recognizes adapters exposing a string model name. Adapters with namespaced identifiers can override this using their canonical model name. Explicit agent settings and explicit model selections take precedence over this hook.

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

def get_default_model_settings(self) -> ModelSettings | None:
    """Return defaults for this model, or None to retain the agent's implicit defaults.

    The default implementation recognizes adapters exposing a string ``model`` name.
    Adapters with namespaced identifiers can override this using their canonical model name.
    Explicit agent settings and explicit model selections take precedence over this hook.
    """
    from .default_models import get_default_model_settings

    name = getattr(self, "model", None)
    return get_default_model_settings(name) if isinstance(name, str) else None

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

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)