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

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

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)