Skip to content

OpenResponses Client

The OpenResponsesClient uses OpenAI's Responses API (POST /v1/responses) instead of the Chat Completions API. This client is useful for providers that implement the newer Responses API format.

Key Differences from ChatCompletionsClient

Feature ChatCompletionsClient OpenResponsesClient
API endpoint chat.completions.create() responses.create()
System messages Included in messages array Passed as instructions parameter
Message format {"role": "user", "content": [...]} {"role": "user", "content": [{"type": "input_text", ...}]}
Tool call IDs tool_call_id call_id
Reasoning config reasoning_effort param reasoning: {"effort": ...} object

Usage

For reasoning models such as the GPT-5.6 family, you can configure the reasoning effort:

import asyncio

from stirrup import Agent
from stirrup.clients import OpenResponsesClient


async def main() -> None:
    """Run an agent using the OpenResponses API with a reasoning model."""

    # Create client using OpenResponsesClient
    # Uses the OpenAI Responses API (responses.create)
    # For reasoning models, you can set reasoning_effort
    client = OpenResponsesClient(
        model="gpt-5.6-luna",
        max_tokens=8_192,
        context_window_tokens=1_000_000,
        reasoning_effort="medium",
    )

    agent = Agent(client=client, name="reasoning-agent", max_turns=19)

    async with agent.session(output_dir="output/open_responses_example") as session:
        _finish_params, _history, _metadata = await session.run(
            "Plan a software release with these tasks: Design (5 days), Backend (10 days, needs Design), "
            "Frontend (8 days, needs Design), Testing (4 days, needs Backend and Frontend), "
            "Documentation (3 days, can start after Backend). Two developers are available. "
            "What's the minimum time to complete? Output an Excel Gantt chart with the schedule."
        )


if __name__ == "__main__":
    asyncio.run(main())

Constructor Parameters

Parameter Type Default Description
model str required Model identifier (e.g., "gpt-5.6-luna", "gpt-5.6-sol")
max_tokens int 64_000 Maximum output tokens
context_window_tokens int required Context capacity for summarization
base_url str \| None None Custom API base URL
api_key str \| None None API key (falls back to OPENAI_API_KEY env var)
reasoning_effort str \| None None Reasoning effort for reasoning models: "low", "medium", "high"
timeout float \| None None Request timeout in seconds
max_retries int 2 Number of retries for transient errors
instructions str \| None None Default system instructions
kwargs dict \| None None Additional arguments passed to responses.create()

API Reference

stirrup.clients.open_responses_client

OpenAI SDK-based LLM client for the Responses API.

This client uses the official OpenAI Python SDK's responses.create() method, supporting both OpenAI's API and any OpenAI-compatible endpoint that implements the Responses API via the base_url parameter.

__all__ module-attribute 

__all__ = ['OpenResponsesClient']

logger module-attribute 

logger = getLogger(__name__)

_OWNED_REQUEST_KEYS module-attribute 

_OWNED_REQUEST_KEYS = frozenset(
    {
        "background",
        "conversation",
        "input",
        "instructions",
        "max_output_tokens",
        "model",
        "previous_response_id",
        "store",
        "stream",
    }
)

AssistantBlock 

AssistantBlock = Annotated[
    TextBlock
    | ReasoningBlock
    | SignedReasoningBlock
    | RedactedReasoningBlock
    | ReasoningRefBlock
    | EncryptedReasoningBlock
    | OpaqueBlock
    | ToolCall
    | ImageContentBlock
    | VideoContentBlock
    | AudioContentBlock,
    Field(discriminator=kind),
]

One block of an assistant turn, discriminated on kind.

ChatMessage 

ChatMessage = Annotated[
    SystemMessage
    | UserRoleMessage
    | AssistantMessage
    | ToolMessage,
    Field(discriminator=role),
]

Discriminated union of all message types, automatically parsed based on role field.

Content 

Content = list[ContentBlock] | str

Message content: either a plain string or list of mixed content blocks.

ContextOverflowError

Bases: Exception

Raised when request input exceeds the model's context capacity.

IncompleteResponseError

Bases: Exception

Raised when a provider returns an incomplete response Stirrup cannot recover from.

Covers stop reasons other than the context and output-budget cases, such as a content filter. Retrying the same request is not expected to help.

OutputTokenLimitError 

OutputTokenLimitError(
    *,
    model_slug: str,
    max_tokens: int,
    provider_reason: str,
)

Bases: Exception

Raised when a provider exhausts the configured response-token budget.

Source code in src/stirrup/core/exceptions.py 
21
22
23
24
25
26
27
28
29
def __init__(self, *, model_slug: str, max_tokens: int, provider_reason: str) -> None:
    self.model_slug = model_slug
    self.max_tokens = max_tokens
    self.provider_reason = provider_reason
    super().__init__(
        f"Model '{model_slug}' exhausted its configured output budget "
        f"(max_tokens={max_tokens}; provider reason: {provider_reason}). "
        "Increase max_tokens or ask the model for a shorter response."
    )

AssistantMessage

Bases: BaseModel

LLM response message: an ordered sequence of assistant blocks.

blocks is the only stored content. The channel-era content and tool_calls attributes remain deprecated views; reasoning raises because an ordered reasoning block sequence has no faithful channel-shaped projection. Serialized v0.1 payloads upgrade to blocks during validation. Channel-shaped construction is not part of the v0.2 API; new code constructs blocks directly. Mixing blocks with non-empty legacy channel keys raises.

provider_response_id class-attribute instance-attribute 

provider_response_id: str | None = None

Provider-attached continuation state, e.g. an OpenAI Responses resp_... id.

This is turn metadata rather than emitted assistant content, so it lives beside blocks instead of inside their emission order. It is distinct from id (Stirrup's message identity) and ReasoningRefBlock.id (an emitted reasoning item handle).

content property 

content: list[AssistantBlock] | str

Bare text for one text block, empty text for no blocks, or the block list.

reasoning property 

reasoning: Reasoning | None

Deprecated channel accessor retained only to fail with migration guidance.

tool_calls property 

tool_calls: list[ToolCall]

Tool calls in emission order.

e2e_otps property 

e2e_otps: float | None

End-to-end output tokens per second.

AudioContentBlock

Bases: BinaryContentBlock

Audio content supporting MPEG, WAV, AAC, and other common audio formats.

to_base64_url 

to_base64_url(bitrate: str = '192k') -> str

Transcode to MP3 and return base64 data URL.

Source code in src/stirrup/core/models.py 
250
251
252
253
254
255
256
257
258
259
260
def to_base64_url(self, bitrate: str = "192k") -> str:
    """Transcode to MP3 and return base64 data URL."""
    with warnings.catch_warnings():
        warnings.filterwarnings("ignore", category=UserWarning, module="moviepy.*")
        with NamedTemporaryFile(suffix=".bin") as fin, NamedTemporaryFile(suffix=".mp3") as fout:
            fin.write(self.data)
            fin.flush()
            clip = AudioFileClip(fin.name)
            clip.write_audiofile(fout.name, codec="libmp3lame", bitrate=bitrate, logger=None)
            clip.close()
            return f"data:audio/mpeg;base64,{b64encode(fout.read()).decode()}"

EmptyParams

Bases: BaseModel

Empty parameter model for tools that don't require parameters.

EncryptedReasoningBlock

Bases: BaseModel

Reasoning returned as an opaque encrypted payload for stateless passback.

E.g. OpenAI Responses reasoning items requested with include: ["reasoning.encrypted_content"] (store=false / zero-data-retention): the item — id, summary parts, and encrypted payload — is re-emitted verbatim in position on passback. The payload is opaque and non-inspectable.

ImageContentBlock

Bases: BinaryContentBlock

Image content supporting PNG, JPEG, WebP, PSD formats with automatic downscaling.

to_base64_url 

to_base64_url(
    max_pixels: int | None = RESOLUTION_1MP,
) -> str

Convert image to base64 data URL, optionally resizing to max pixel count.

Source code in src/stirrup/core/models.py 
157
158
159
160
161
162
163
164
165
166
167
def to_base64_url(self, max_pixels: int | None = RESOLUTION_1MP) -> str:
    """Convert image to base64 data URL, optionally resizing to max pixel count."""
    img: Image.Image = Image.open(BytesIO(self.data))
    if max_pixels is not None and img.width * img.height > max_pixels:
        tw, th = downscale_image(img.width, img.height, max_pixels)
        img.thumbnail((tw, th), Image.Resampling.LANCZOS)
    if img.mode != "RGB":
        img = img.convert("RGB")
    buf = BytesIO()
    img.save(buf, format="PNG")
    return f"data:image/png;base64,{b64encode(buf.getvalue()).decode()}"

LLMClient

Bases: Protocol

Protocol defining the interface for LLM client implementations.

Any LLM client must implement this protocol to work with the Agent class. Provides text generation with tool support and model capability inspection.

OpaqueBlock

Bases: BaseModel

Provider-native block the framework carries uninterpreted.

For provider-issued marker/control blocks that must round-trip untouched: data holds the block's raw JSON (self-describing — the provider's own type field travels inside it). The framework preserves it in position through history, projections, and serialization so a client that understands the payload can re-emit it verbatim on passback; other clients fail loudly.

ReasoningBlock

Bases: BaseModel

In-band reasoning text with no passback token.

E.g. reasoning_content on Chat Completions-compatible hosts, or -tag extraction.

ReasoningRefBlock

Bases: BaseModel

Reasoning held provider-side and passed back by reference.

This is retained for providers that require an item-level handle on replay. OpenAI Responses continuation uses AssistantMessage.provider_response_id instead and does not create this block.

RedactedReasoningBlock

Bases: BaseModel

Reasoning the provider withheld, replaced by an opaque payload.

E.g. Anthropic redacted_thinking: data must be re-emitted verbatim as a redacted_thinking block on passback. Carries no readable content.

SignedReasoningBlock

Bases: BaseModel

Reasoning bound to an opaque provider signature re-emitted verbatim on passback.

E.g. Anthropic signed thinking blocks.

SystemMessage

Bases: BaseModel

System-level instructions and context for the LLM.

TextBlock

Bases: BaseModel

One contiguous run of answer text in an assistant turn.

signature carries opaque passback state attached to this exact block, e.g. a Google thought signature emitted on a visible text part. A client that cannot re-emit the signature must reject passback rather than silently stripping it.

TokenUsage

Bases: BaseModel

Token counts for LLM usage.

Token terminology: output = reasoning + answer.

output property 

output: int

Total output tokens (reasoning + answer).

total property 

total: int

Total token count across input, answer, and reasoning.

__add__ 

__add__(other: TokenUsage) -> TokenUsage

Add two TokenUsage objects together, summing each field independently.

Source code in src/stirrup/core/models.py 
479
480
481
482
483
484
485
def __add__(self, other: "TokenUsage") -> "TokenUsage":
    """Add two TokenUsage objects together, summing each field independently."""
    return TokenUsage(
        input=self.input + other.input,
        answer=self.answer + other.answer,
        reasoning=self.reasoning + other.reasoning,
    )

Tool

Bases: BaseModel

Tool definition with name, description, parameter schema, and executor function.

Generic over

P: Parameter model type (Pydantic BaseModel subclass, or EmptyParams for parameterless tools) M: Metadata type (should implement Addable for aggregation; use None for tools without metadata)

Tools are simple, stateless callables. For tools requiring lifecycle management (setup/teardown, resource pooling), use a ToolProvider instead.

Example with parameters 
class CalcParams(BaseModel):
    expression: str

calc_tool = Tool[CalcParams, None](
    name="calc",
    description="Evaluate math",
    parameters=CalcParams,
    executor=lambda p: ToolResult(content=str(eval(p.expression))),
)

Example without parameters (uses EmptyParams by default): 

time_tool = Tool[EmptyParams, None](
    name="time",
    description="Get current time",
    executor=lambda _: ToolResult(content=datetime.now().isoformat()),
)

ToolCall

Bases: BaseModel

Represents a tool invocation request from the LLM.

Also a member of the AssistantBlock union: the kind discriminator is defaulted, so legacy payloads without the key still validate anywhere ToolCall is used as a plain input, and new dumps always carry it.

Attributes:

Name Type Description
name str

Name of the tool to invoke
arguments str

JSON string containing tool parameters
tool_call_id str

Unique identifier for tracking this tool call and its result

signature class-attribute instance-attribute 

signature: str | None = None

Opaque passback state attached to this exact block, e.g. a Google thought signature.

has_provider_tool_call_id class-attribute instance-attribute 

has_provider_tool_call_id: bool = True

Whether tool_call_id was present on the provider's original block.

A client may synthesize tool_call_id for internal call/result matching while retaining that it must be omitted from provider-attached passback.

from_provider classmethod 

from_provider(
    *,
    provider_id: str | None,
    name: str,
    arguments: str,
    signature: str | None = None,
) -> Self

Capture one provider call with a stable internal correlation ID.

Source code in src/stirrup/core/models.py 
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
@classmethod
def from_provider(
    cls,
    *,
    provider_id: str | None,
    name: str,
    arguments: str,
    signature: str | None = None,
) -> Self:
    """Capture one provider call with a stable internal correlation ID."""
    native_id = provider_id or None
    return cls(
        tool_call_id=native_id or uuid4().hex,
        has_provider_tool_call_id=native_id is not None,
        name=name,
        arguments=arguments,
        signature=signature,
    )

ToolMessage

Bases: BaseModel

Tool execution result returned to the LLM.

Attributes:

Name Type Description
role Literal['tool']

Always "tool"
content Content

The tool result content
tool_call_id str

ID linking this result to the corresponding tool call
name str | None

Name of the tool that was called
args_was_valid bool

Whether the tool arguments were valid
success bool

Whether the tool executed successfully (used by finish tool to control termination)

tool_duration property 

tool_duration: float | None

Tool execution duration in seconds.

UserMessage

Bases: BaseModel

User input message to the LLM.

VideoContentBlock

Bases: BinaryContentBlock

MP4 video content with automatic transcoding and resolution downscaling.

to_base64_url 

to_base64_url(
    max_pixels: int | None = RESOLUTION_480P,
    fps: int | None = None,
) -> str

Transcode to MP4 and return base64 data URL.

Source code in src/stirrup/core/models.py 
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
def to_base64_url(self, max_pixels: int | None = RESOLUTION_480P, fps: int | None = None) -> str:
    """Transcode to MP4 and return base64 data URL."""
    with warnings.catch_warnings():
        warnings.filterwarnings("ignore", category=UserWarning, module="moviepy.*")
        with NamedTemporaryFile(suffix=".mp4") as fin, NamedTemporaryFile(suffix=".mp4") as fout:
            fin.write(self.data)
            fin.flush()
            clip = VideoFileClip(fin.name)
            tw, th = downscale_image(int(clip.w), int(clip.h), max_pixels)
            clip = clip.with_effects([Resize(new_size=(tw, th))])

            clip.write_videofile(
                fout.name,
                codec="libx264",
                fps=fps,
                audio=clip.audio is not None,
                audio_codec="aac",
                preset="veryfast",
                logger=None,
            )
            clip.close()
            return f"data:video/mp4;base64,{b64encode(fout.read()).decode()}"

OpenResponsesClient 

OpenResponsesClient(
    model: str,
    max_tokens: int = 64000,
    *,
    context_window_tokens: int,
    base_url: str | None = None,
    api_key: str | None = None,
    reasoning_effort: str | None = None,
    encrypted_reasoning: bool = False,
    timeout: float | None = None,
    max_retries: int = 2,
    instructions: str | None = None,
    kwargs: dict[str, Any] | None = None,
)

Bases: LLMClient

OpenAI SDK-based client using the Responses API.

Uses the official OpenAI Python SDK's responses.create() method. Supports custom base_url for OpenAI-compatible providers that implement the Responses API.

Delegates retries for transient failures to the OpenAI SDK and tracks token usage.

Example

Standard OpenAI usage

client = OpenResponsesClient( ... model="gpt-5.6-luna", ... max_tokens=8_192, ... context_window_tokens=1_000_000, ... )

Custom OpenAI-compatible endpoint

client = OpenResponsesClient( ... model="gpt-5.6-luna", ... context_window_tokens=1_000_000, ... base_url="http://localhost:8000/v1", ... api_key="your-api-key", ... )

Initialize OpenAI SDK client with model configuration for Responses API.

Parameters:

Name Type Description Default
model str

Model identifier (e.g., 'gpt-5.6-luna', 'gpt-5.6-sol').

 required
max_tokens int

Maximum output tokens. Defaults to 64,000.

 64000
context_window_tokens int

Context capacity used to decide when Agent history should be summarized.

 required
base_url str | None

API base URL. If None, uses OpenAI's standard URL. Use for OpenAI-compatible providers.

 None
api_key str | None

API key for authentication. If None, reads from OPENROUTER_API_KEY environment variable.

 None
reasoning_effort str | None

Reasoning effort level for extended thinking models (e.g., 'low', 'medium', 'high'). Only used with reasoning models.

 None
encrypted_reasoning bool

Run stateless (store=false) and request include: ["reasoning.encrypted_content"] so reasoning items are captured as EncryptedReasoningBlock and re-emitted verbatim in position on passback. Required for zero-data-retention setups where the provider holds no conversation state.

 False
timeout float | None

Request timeout in seconds. If None, uses OpenAI SDK default.

 None
max_retries int

Number of retries for transient errors. Defaults to 2. The OpenAI SDK handles retries internally.

 2
instructions str | None

Default system-level instructions. Can be overridden by SystemMessage in the messages list.

 None
kwargs dict[str, Any] | None

Additional arguments passed to responses.create(). Structural request keys owned by this client are rejected. tools / tool_choice and reasoning remain available for provider-native configuration, but cannot be combined with the corresponding dedicated generate(tools=...) or reasoning_effort configuration.

 None

Raises:

Type Description
ValueError

If context_window_tokens is not positive, max_tokens exceeds it, or kwargs contains a request key owned by this client.
Source code in src/stirrup/clients/open_responses_client.py 
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
def __init__(
    self,
    model: str,
    max_tokens: int = 64_000,
    *,
    context_window_tokens: int,
    base_url: str | None = None,
    api_key: str | None = None,
    reasoning_effort: str | None = None,
    encrypted_reasoning: bool = False,
    timeout: float | None = None,
    max_retries: int = 2,
    instructions: str | None = None,
    kwargs: dict[str, Any] | None = None,
) -> None:
    """Initialize OpenAI SDK client with model configuration for Responses API.

    Args:
        model: Model identifier (e.g., 'gpt-5.6-luna', 'gpt-5.6-sol').
        max_tokens: Maximum output tokens. Defaults to 64,000.
        context_window_tokens: Context capacity used to decide when Agent history
            should be summarized.
        base_url: API base URL. If None, uses OpenAI's standard URL.
            Use for OpenAI-compatible providers.
        api_key: API key for authentication. If None, reads from OPENROUTER_API_KEY
            environment variable.
        reasoning_effort: Reasoning effort level for extended thinking models
            (e.g., 'low', 'medium', 'high'). Only used with reasoning models.
        encrypted_reasoning: Run stateless (``store=false``) and request
            ``include: ["reasoning.encrypted_content"]`` so reasoning items are
            captured as EncryptedReasoningBlock and re-emitted verbatim in
            position on passback. Required for zero-data-retention setups where
            the provider holds no conversation state.
        timeout: Request timeout in seconds. If None, uses OpenAI SDK default.
        max_retries: Number of retries for transient errors. Defaults to 2.
            The OpenAI SDK handles retries internally.
        instructions: Default system-level instructions. Can be overridden by
            SystemMessage in the messages list.
        kwargs: Additional arguments passed to responses.create(). Structural
            request keys owned by this client are rejected. ``tools`` /
            ``tool_choice`` and ``reasoning`` remain available for provider-native
            configuration, but cannot be combined with the corresponding dedicated
            ``generate(tools=...)`` or ``reasoning_effort`` configuration.

    Raises:
        ValueError: If ``context_window_tokens`` is not positive, ``max_tokens``
            exceeds it, or ``kwargs`` contains a request key owned by this client.
    """
    validate_token_budgets(max_tokens, context_window_tokens)

    self._model = model
    self._max_tokens = max_tokens
    self._context_window_tokens = context_window_tokens
    self._reasoning_effort = reasoning_effort
    self._encrypted_reasoning = encrypted_reasoning
    self._default_instructions = instructions
    self._kwargs = dict(kwargs or {})
    reserved_keys = sorted(_OWNED_REQUEST_KEYS & self._kwargs.keys())
    if reserved_keys:
        raise ValueError(
            f"OpenResponsesClient owns request keys {reserved_keys}; use its dedicated arguments and message "
            "history instead of kwargs"
        )

    # Initialize AsyncOpenAI client
    resolved_api_key = api_key or os.environ.get("OPENAI_API_KEY")

    # Strip /responses suffix if present - SDK appends it automatically
    resolved_base_url = base_url
    if resolved_base_url and resolved_base_url.rstrip("/").endswith("/responses"):
        resolved_base_url = resolved_base_url.rstrip("/").removesuffix("/responses")

    self._client = AsyncOpenAI(
        api_key=resolved_api_key,
        base_url=resolved_base_url,
        timeout=timeout,
        max_retries=max_retries,
    )

max_tokens property 

max_tokens: int

Maximum output tokens.

context_window_tokens property 

context_window_tokens: int

Context capacity used by agents for history summarization.

model_slug property 

model_slug: str

Model identifier.

generate async 

generate(
    messages: list[ChatMessage], tools: dict[str, Tool]
) -> AssistantMessage

Generate assistant response with optional tool calls using Responses API.

Parameters:

Name Type Description Default
messages list[ChatMessage]

List of conversation messages.

 required
tools dict[str, Tool]

Dictionary mapping tool names to Tool objects.

 required

Returns:

Type Description
AssistantMessage

AssistantMessage containing the model's response, any tool calls,
AssistantMessage

and token usage statistics.

Raises:

Type Description
ContextOverflowError

If the provider rejects the request because the input exceeds the model's context capacity.
OutputTokenLimitError

If the provider exhausts max_tokens.
IncompleteResponseError

If the response is incomplete for another provider reason, such as a content filter.
Source code in src/stirrup/clients/open_responses_client.py 
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
async def generate(
    self,
    messages: list[ChatMessage],
    tools: dict[str, Tool],
) -> AssistantMessage:
    """Generate assistant response with optional tool calls using Responses API.

    Args:
        messages: List of conversation messages.
        tools: Dictionary mapping tool names to Tool objects.

    Returns:
        AssistantMessage containing the model's response, any tool calls,
        and token usage statistics.

    Raises:
        ContextOverflowError: If the provider rejects the request because the input
            exceeds the model's context capacity.
        OutputTokenLimitError: If the provider exhausts ``max_tokens``.
        IncompleteResponseError: If the response is incomplete for another provider
            reason, such as a content filter.
    """
    conditional_keys: set[str] = set()
    if tools:
        conditional_keys.update(("tools", "tool_choice"))
    if self._reasoning_effort:
        conditional_keys.add("reasoning")
    if conflicts := sorted(conditional_keys & self._kwargs.keys()):
        raise ValueError(f"OpenResponsesClient kwargs keys {conflicts} conflict with dedicated call configuration")

    # ``encrypted_reasoning`` owns statefulness; kwargs cannot override ``store``.
    stateful = not self._encrypted_reasoning

    # Convert messages to OpenResponses format. Stateful calls continue from the
    # latest provider response; stateless/ZDR calls replay encrypted reasoning items.
    instructions, previous_response_id, input_items = _to_open_responses_request(
        messages,
        use_provider_response_id=stateful,
    )

    # Use provided instructions or fall back to default
    final_instructions = instructions or self._default_instructions

    # Build request kwargs
    request_kwargs: dict[str, Any] = {
        **self._kwargs,
        "model": self._model,
        "input": input_items,
        "max_output_tokens": self._max_tokens,
    }
    if previous_response_id is not None:
        request_kwargs["previous_response_id"] = previous_response_id

    # Add instructions if present
    if final_instructions:
        request_kwargs["instructions"] = final_instructions

    # Add tools if provided
    if tools:
        request_kwargs["tools"] = _to_open_responses_tools(tools)
        request_kwargs["tool_choice"] = "auto"

    # Add reasoning effort if configured (for reasoning models)
    if self._reasoning_effort:
        request_kwargs["reasoning"] = {"effort": self._reasoning_effort}

    # Stateless mode: nothing is stored provider-side, so ask for the encrypted
    # reasoning payload and carry it client-side across turns. A caller-supplied
    # include list (via kwargs) is extended, not clobbered.
    if self._encrypted_reasoning:
        request_kwargs["store"] = False
        include = list(request_kwargs.get("include") or [])
        if "reasoning.encrypted_content" not in include:
            include.append("reasoning.encrypted_content")
        request_kwargs["include"] = include

    # Make API call
    request_start_time = perf_counter()
    try:
        response = await self._create_response(request_kwargs)
    except APIStatusError as error:
        # BadRequestError subclasses APIStatusError, so both provider signals are matched
        # in this one handler; a separate `except BadRequestError` would be dead code for
        # whichever signal the other clause re-raises. Only OpenAI's own context code is
        # recognised. Compatible endpoints that report a different code surface as a bad
        # request instead of being recovered; widening this trades that for false positives.
        if error.code == "context_length_exceeded":
            raise ContextOverflowError(str(error)) from error
        if previous_response_id is None or not _is_missing_previous_response(error):
            raise
        try:
            replay_instructions, _, replay_input = _to_open_responses_request(
                messages,
                use_provider_response_id=False,
            )
        except (NotImplementedError, TypeError) as replay_error:
            raise RuntimeError(
                "The stored OpenAI response is unavailable and this history cannot be replayed exactly "
                f"without its provider-side continuation state: {replay_error}"
            ) from error

        logger.warning(
            "Stored OpenAI response %s was not found; retrying once with a full local-history replay",
            previous_response_id,
        )
        request_kwargs["input"] = replay_input
        request_kwargs.pop("previous_response_id", None)
        final_replay_instructions = replay_instructions or self._default_instructions
        if final_replay_instructions:
            request_kwargs["instructions"] = final_replay_instructions
        else:
            request_kwargs.pop("instructions", None)
        # The replay sends the whole local history where the original sent a short
        # pointer, so it is the likeliest point to exceed the window. Map that to the
        # recoverable error too, rather than ending the run with a raw bad request.
        try:
            response = await self._create_response(request_kwargs)
        except APIStatusError as replay_overflow:
            if replay_overflow.code != "context_length_exceeded":
                raise
            raise ContextOverflowError(str(replay_overflow)) from replay_overflow
    request_end_time = perf_counter()

    if response.status == "incomplete":
        incomplete_details = getattr(response, "incomplete_details", None)
        incomplete_reason = getattr(incomplete_details, "reason", None) or "unknown"
        if incomplete_reason == "max_output_tokens":
            raise OutputTokenLimitError(
                model_slug=self.model_slug,
                max_tokens=self._max_tokens,
                provider_reason=incomplete_reason,
            )
        raise IncompleteResponseError(f"Response incomplete for model {self.model_slug}: {incomplete_reason}")
    if response.status != "completed":
        raise RuntimeError(
            f"OpenAI Responses returned unsupported status {response.status!r}: "
            f"{getattr(response, 'error', None)!r}"
        )

    provider_response_id: str | None = None
    if stateful:
        response_id = response.id
        if not isinstance(response_id, str) or not response_id:
            raise ValueError("Stored OpenAI Responses calls require a non-empty response ID")
        provider_response_id = response_id

    # Parse response output into ordered blocks
    blocks = _parse_response_output(response.output, allow_reference_reasoning=stateful)

    # Parse token usage
    usage = response.usage
    input_tokens = usage.input_tokens if usage else 0
    output_tokens = usage.output_tokens if usage else 0

    # Handle reasoning tokens if available
    reasoning_tokens = 0
    if usage and hasattr(usage, "output_tokens_details") and usage.output_tokens_details:
        reasoning_tokens = getattr(usage.output_tokens_details, "reasoning_tokens", 0) or 0

    answer_tokens = output_tokens - reasoning_tokens

    return AssistantMessage(
        provider_response_id=provider_response_id,
        blocks=blocks,
        token_usage=TokenUsage(
            input=input_tokens,
            answer=answer_tokens,
            reasoning=reasoning_tokens,
        ),
        request_start_time=request_start_time,
        request_end_time=request_end_time,
    )

validate_token_budgets 

validate_token_budgets(
    max_tokens: int, context_window_tokens: int
) -> None

Reject an invalid budget pair at client construction.

Raises:

Type Description
ValueError

If context_window_tokens is not positive, or max_tokens exceeds it.
Source code in src/stirrup/clients/utils.py 
45
46
47
48
49
50
51
52
53
54
55
def validate_token_budgets(max_tokens: int, context_window_tokens: int) -> None:
    """Reject an invalid budget pair at client construction.

    Raises:
        ValueError: If ``context_window_tokens`` is not positive, or
            ``max_tokens`` exceeds it.
    """
    if context_window_tokens <= 0:
        raise ValueError(f"context_window_tokens must be positive, got {context_window_tokens!r}")
    if max_tokens > context_window_tokens:
        raise ValueError(f"max_tokens ({max_tokens}) must not exceed context_window_tokens ({context_window_tokens})")

_is_missing_previous_response 

_is_missing_previous_response(
    error: APIStatusError,
) -> bool

Match only the provider error that identifies an unavailable continuation.

Source code in src/stirrup/clients/open_responses_client.py 
65
66
67
def _is_missing_previous_response(error: APIStatusError) -> bool:
    """Match only the provider error that identifies an unavailable continuation."""
    return error.code == "previous_response_not_found" and error.param == "previous_response_id"

_content_to_open_responses_input 

_content_to_open_responses_input(
    content: Content,
) -> list[dict[str, Any]]

Convert Content blocks to OpenResponses input content format.

Uses input_text for text content (vs output_text for responses).

Source code in src/stirrup/clients/open_responses_client.py 
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
def _content_to_open_responses_input(content: Content) -> list[dict[str, Any]]:
    """Convert Content blocks to OpenResponses input content format.

    Uses input_text for text content (vs output_text for responses).
    """
    if isinstance(content, str):
        return [{"type": "input_text", "text": content}]

    out: list[dict[str, Any]] = []
    for block in content:
        if isinstance(block, str):
            out.append({"type": "input_text", "text": block})
        elif isinstance(block, ImageContentBlock):
            out.append({"type": "input_image", "image_url": block.to_base64_url()})
        elif isinstance(block, AudioContentBlock):
            out.append(
                {
                    "type": "input_audio",
                    "input_audio": {
                        "data": block.to_base64_url().split(",")[1],
                        "format": block.extension,
                    },
                }
            )
        elif isinstance(block, VideoContentBlock):
            out.append({"type": "input_file", "file_data": block.to_base64_url()})
        else:
            raise NotImplementedError(f"Unsupported content block: {type(block)}")
    return out

_content_to_open_responses_output 

_content_to_open_responses_output(
    content: Content,
) -> list[dict[str, Any]]

Convert Content blocks to OpenResponses output content format.

Uses output_text for assistant message content.

Source code in src/stirrup/clients/open_responses_client.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
def _content_to_open_responses_output(content: Content) -> list[dict[str, Any]]:
    """Convert Content blocks to OpenResponses output content format.

    Uses output_text for assistant message content.
    """
    if isinstance(content, str):
        return [{"type": "output_text", "text": content}]

    out: list[dict[str, Any]] = []
    for block in content:
        if isinstance(block, str):
            out.append({"type": "output_text", "text": block})
        else:
            raise NotImplementedError(f"Unsupported output content block: {type(block)}")
    return out

_to_open_responses_tools 

_to_open_responses_tools(
    tools: dict[str, Tool],
) -> list[dict[str, Any]]

Convert Tool objects to OpenResponses function format.

OpenResponses API expects tools with name/description/parameters at top level, not nested under a 'function' key like Chat Completions API.

Parameters:

Name Type Description Default
tools dict[str, Tool]

Dictionary mapping tool names to Tool objects.

 required

Returns:

Type Description
list[dict[str, Any]]

List of tool definitions in OpenResponses format.
Source code in src/stirrup/clients/open_responses_client.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def _to_open_responses_tools(tools: dict[str, Tool]) -> list[dict[str, Any]]:
    """Convert Tool objects to OpenResponses function format.

    OpenResponses API expects tools with name/description/parameters at top level,
    not nested under a 'function' key like Chat Completions API.

    Args:
        tools: Dictionary mapping tool names to Tool objects.

    Returns:
        List of tool definitions in OpenResponses format.
    """
    out: list[dict[str, Any]] = []
    for t in tools.values():
        tool_def: dict[str, Any] = {
            "type": "function",
            "name": t.name,
            "description": t.description,
        }
        if t.parameters is not EmptyParams:
            tool_def["parameters"] = t.parameters.model_json_schema()
        out.append(tool_def)
    return out

_to_open_responses_input 

_to_open_responses_input(
    msgs: Sequence[ChatMessage],
    *,
    allow_reference_reasoning: bool = True,
) -> tuple[str | None, list[dict[str, Any]]]

Convert ChatMessage list to OpenResponses (instructions, input) tuple.

SystemMessage content is extracted as the instructions parameter. Other messages are converted to input items.

Returns:

Type Description
str | None

Tuple of (instructions, input_items) where instructions is the system
list[dict[str, Any]]

message content (or None) and input_items is the list of input items.
Source code in src/stirrup/clients/open_responses_client.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
def _to_open_responses_input(
    msgs: Sequence[ChatMessage],
    *,
    allow_reference_reasoning: bool = True,
) -> tuple[str | None, list[dict[str, Any]]]:
    """Convert ChatMessage list to OpenResponses (instructions, input) tuple.

    SystemMessage content is extracted as the instructions parameter.
    Other messages are converted to input items.

    Returns:
        Tuple of (instructions, input_items) where instructions is the system
        message content (or None) and input_items is the list of input items.
    """
    instructions: str | None = None
    input_items: list[dict[str, Any]] = []

    for m in msgs:
        if isinstance(m, SystemMessage):
            # Extract system message as instructions
            if isinstance(m.content, str):
                instructions = m.content
            else:
                # Join text content blocks for instructions
                instructions = "\n".join(block if isinstance(block, str) else "" for block in m.content)
        elif isinstance(m, UserMessage):
            input_items.append(
                {
                    "role": "user",
                    "content": _content_to_open_responses_input(m.content),
                }
            )
        elif isinstance(m, AssistantMessage):
            # Assistant turns replay as response output items, one item per block,
            # in the model's true emission order (message / function_call / reasoning
            # interleaved exactly as captured). Channel-synthesized (legacy) messages
            # carry blocks in channel order, so they replay in the old
            # message-then-calls order automatically.
            for block in m.blocks:
                match block:
                    case TextBlock(text=text, signature=None):
                        input_items.append(
                            {
                                "type": "message",
                                "role": "assistant",
                                "content": [{"type": "output_text", "text": text}],
                            }
                        )
                    case TextBlock():
                        raise TypeError("OpenAI Responses cannot pass back signed text blocks")
                    case ToolCall(
                        tool_call_id=call_id,
                        name=name,
                        arguments=arguments,
                        signature=None,
                        has_provider_tool_call_id=True,
                    ):
                        input_items.append(
                            {
                                "type": "function_call",
                                "call_id": call_id,
                                "name": name,
                                "arguments": arguments,
                            }
                        )
                    case ToolCall():
                        raise TypeError("OpenAI Responses cannot pass back provider-attached or synthetic tool calls")
                    case ReasoningRefBlock() if not allow_reference_reasoning:
                        raise NotImplementedError(
                            "Stateless OpenAI Responses passback requires encrypted reasoning content"
                        )
                    case ReasoningRefBlock(
                        id=reasoning_id,
                        content=content,
                        summary=summary,
                    ):
                        # Passed back by reference: the id is the handle to the stored
                        # reasoning item.
                        reasoning_item: dict[str, Any] = {
                            "type": "reasoning",
                            "id": reasoning_id,
                            "content": [] if content is None else [{"type": "reasoning_text", "text": content}],
                            "summary": [{"type": "summary_text", "text": part} for part in summary],
                        }
                        input_items.append(reasoning_item)
                    case EncryptedReasoningBlock(
                        id=reasoning_id,
                        content=content,
                        summary=summary,
                        encrypted_content=encrypted_content,
                    ):
                        # Stateless passback: the whole item — id, summary parts, and
                        # encrypted payload — is re-emitted verbatim in position.
                        input_items.append(
                            {
                                "type": "reasoning",
                                "id": reasoning_id,
                                "content": [] if content is None else [{"type": "reasoning_text", "text": content}],
                                "summary": [{"type": "summary_text", "text": part} for part in summary],
                                "encrypted_content": encrypted_content,
                            }
                        )
                    case ReasoningBlock():
                        raise NotImplementedError(
                            "OpenAI Responses cannot pass back reasoning without an item ID or encrypted content"
                        )
                    case (
                        SignedReasoningBlock()
                        | RedactedReasoningBlock()
                        | OpaqueBlock()
                        | ImageContentBlock()
                        | VideoContentBlock()
                        | AudioContentBlock()
                    ):
                        raise NotImplementedError(
                            f"OpenAI Responses cannot pass back {type(block).__name__} assistant blocks"
                        )
                    case _:
                        assert_never(block)
        elif isinstance(m, ToolMessage):
            # Tool results are function_call_output items
            if not isinstance(m.content, str):
                raise NotImplementedError(
                    "OpenAI Responses tool outputs currently require string content; multimodal tool output "
                    "needs a lossless provider mapping"
                )
            input_items.append(
                {
                    "type": "function_call_output",
                    "call_id": m.tool_call_id,
                    "output": m.content,
                }
            )
        else:
            raise NotImplementedError(f"Unsupported message type: {type(m)}")

    return instructions, input_items

_to_open_responses_request 

_to_open_responses_request(
    msgs: Sequence[ChatMessage],
    *,
    use_provider_response_id: bool,
) -> tuple[str | None, str | None, list[dict[str, Any]]]

Build one Responses request while preserving response-level continuation state.

previous_response_id replaces only the earlier input items. System instructions are extracted from the complete local history because the Responses API does not carry a previous response's instructions into the next request.

Source code in src/stirrup/clients/open_responses_client.py 
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
def _to_open_responses_request(
    msgs: Sequence[ChatMessage],
    *,
    use_provider_response_id: bool,
) -> tuple[str | None, str | None, list[dict[str, Any]]]:
    """Build one Responses request while preserving response-level continuation state.

    ``previous_response_id`` replaces only the earlier input items. System instructions
    are extracted from the complete local history because the Responses API does not
    carry a previous response's ``instructions`` into the next request.
    """
    instructions, _ = _to_open_responses_input([msg for msg in msgs if isinstance(msg, SystemMessage)])
    previous_response_id: str | None = None
    messages_to_process = msgs

    if use_provider_response_id:
        for index in range(len(msgs) - 1, -1, -1):
            msg = msgs[index]
            if isinstance(msg, AssistantMessage) and msg.provider_response_id is not None:
                if any(isinstance(block, ToolCall) and not block.has_provider_tool_call_id for block in msg.blocks):
                    raise NotImplementedError(
                        "Stored OpenAI Responses continuation cannot correlate a provider-idless tool call"
                    )
                previous_response_id = msg.provider_response_id
                messages_to_process = msgs[index + 1 :]
                break

    _, input_items = _to_open_responses_input(
        messages_to_process,
        allow_reference_reasoning=previous_response_id is not None,
    )
    return instructions, previous_response_id, input_items

_get_attr 

_get_attr(obj: Any, name: str, default: Any = None) -> Any

Get attribute from object or dict, with fallback default.

Source code in src/stirrup/clients/open_responses_client.py 
316
317
318
319
320
def _get_attr(obj: Any, name: str, default: Any = None) -> Any:  # noqa: ANN401
    """Get attribute from object or dict, with fallback default."""
    if isinstance(obj, dict):
        return obj.get(name, default)
    return getattr(obj, name, default)

_parse_response_output 

_parse_response_output(
    output: list[Any],
    *,
    allow_reference_reasoning: bool = True,
) -> list[AssistantBlock]

Parse response output items into ordered assistant blocks.

One exhaustive pass in item order: each message item becomes its own TextBlock (refusal content surfaces as answer text), each function_call a ToolCall block, and each reasoning item an EncryptedReasoningBlock or readable ReasoningBlock. Stored continuation state lives on the assistant message as provider_response_id; reasoning item IDs are not duplicated into blocks. Unknown item and content types raise until their semantics and passback behavior are explicitly implemented.

Source code in src/stirrup/clients/open_responses_client.py 
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
def _parse_response_output(
    output: list[Any],
    *,
    allow_reference_reasoning: bool = True,
) -> list[AssistantBlock]:
    """Parse response output items into ordered assistant blocks.

    One exhaustive pass in item order: each ``message`` item becomes its own
    TextBlock (refusal content surfaces as answer text), each ``function_call``
    a ToolCall block, and each ``reasoning`` item an EncryptedReasoningBlock or
    readable ReasoningBlock. Stored continuation state lives on the assistant
    message as ``provider_response_id``; reasoning item IDs are not duplicated
    into blocks. Unknown item and content types raise until their semantics and
    passback behavior are explicitly implemented.
    """
    blocks: list[AssistantBlock] = []

    for item in output:
        item_type = _get_attr(item, "type")

        match item_type:
            case "message":
                text_parts: list[str] = []
                for content_item in _get_attr(item, "content", []):
                    content_type = _get_attr(content_item, "type")
                    if content_type == "refusal":
                        # A refusal is a normal API response, not malformed output: surface
                        # its text as the assistant's answer rather than crashing the turn.
                        refusal = _get_attr(content_item, "refusal")
                        if not isinstance(refusal, str):
                            raise ValueError(f"OpenAI Responses refusal is not a string: {refusal!r}")
                        text_parts.append(refusal)
                        continue
                    if content_type != "output_text":
                        raise NotImplementedError(
                            f"Unsupported OpenAI Responses message content type: {content_type!r}"
                        )
                    text = _get_attr(content_item, "text")
                    if not isinstance(text, str):
                        raise ValueError(f"OpenAI Responses output_text is not a string: {text!r}")
                    text_parts.append(text)
                if text_parts:
                    blocks.append(TextBlock(text="".join(text_parts)))

            case "function_call":
                blocks.append(
                    ToolCall.from_provider(
                        provider_id=_get_attr(item, "call_id"),
                        name=_get_attr(item, "name"),
                        arguments=_get_attr(item, "arguments", ""),
                    )
                )

            case "reasoning":
                content = _get_attr(item, "content")
                content_parts: list[str] = []
                if isinstance(content, list):
                    for part in content:
                        text = _get_attr(part, "text")
                        if not isinstance(text, str):
                            raise ValueError(f"OpenAI Responses reasoning content entry has no text: {part!r}")
                        if text:
                            content_parts.append(text)
                elif content:
                    content_parts = [str(content)]

                # summary can be a list of Summary objects with .text attribute
                summary = _get_attr(item, "summary")
                if isinstance(summary, list):
                    summary_parts = []
                    for s in summary:
                        text = _get_attr(s, "text")
                        if not isinstance(text, str):
                            raise ValueError(f"OpenAI Responses reasoning summary entry has no text: {s!r}")
                        # Empty parts carry no data; dropping them keeps re-emission faithful.
                        if text:
                            summary_parts.append(text)
                elif summary:
                    summary_parts = [str(summary)]
                else:
                    thinking = _get_attr(item, "thinking")
                    summary_parts = [thinking] if thinking else []

                item_id = _get_attr(item, "id")
                encrypted_content = _get_attr(item, "encrypted_content")
                if encrypted_content and not item_id:
                    # The id is the passback handle; without it the encrypted payload
                    # cannot be re-emitted and ZDR reasoning state would silently vanish.
                    raise ValueError(f"OpenAI Responses reasoning item has encrypted_content but no id: {item!r}")
                if item_id and encrypted_content:
                    # Stateless (store=false / ZDR) item: carried whole for verbatim re-emission.
                    blocks.append(
                        EncryptedReasoningBlock(
                            id=item_id,
                            encrypted_content=encrypted_content,
                            content="".join(content_parts) or None,
                            summary=summary_parts,
                        )
                    )
                elif item_id:
                    if not allow_reference_reasoning:
                        raise NotImplementedError(
                            "A stateless OpenAI Responses call returned reference-only reasoning; "
                            "enable encrypted_reasoning so it can be passed back"
                        )
                    readable_parts = [*content_parts, *summary_parts]
                    if readable_parts:
                        blocks.append(ReasoningBlock(content="\n".join(readable_parts)))
                elif content_parts or summary_parts:
                    blocks.append(ReasoningBlock(content="\n".join([*content_parts, *summary_parts])))

            case _:
                raise NotImplementedError(f"Unsupported OpenAI Responses output item type: {item_type!r}")

    return blocks