axio-responses

The OpenAI Responses API as axio speaks it: request items in, StreamEvents out.

Both halves live here rather than in a transport because two transports speak this API — the public /v1/responses endpoint and the ChatGPT backend Codex uses. The package knows nothing about HTTP and opens no connection.

import json

from axio import Message, ReasoningBlock, TextBlock
from axio_responses import Responses, convert_messages
from axio_sse import Event

# A stored reasoning block goes back as its own item. Without the encrypted content and
# the id the model starts the next round blind, so a block that has neither is dropped.
instructions, items = convert_messages(
    [
        Message(role="user", content=[TextBlock(text="Why?")]),
        Message(
            role="assistant",
            content=[
                ReasoningBlock(text="", signature="gAAAAA...", id="rs_1"),
                TextBlock(text="Because."),
            ],
        ),
    ],
    "You are helpful.",
)
assert instructions == "You are helpful."
assert items[1] == {
    "type": "reasoning",
    "id": "rs_1",
    "encrypted_content": "gAAAAA...",
    "summary": [],
}

reader = Responses()


def read(**payload: object) -> list[object]:
    return list(reader.read(Event(data=json.dumps(payload))))


[delta] = read(type="response.output_text.delta", delta="Hello")
assert delta.delta == "Hello"

# A hosted tool the reader does not name, forwarded rather than dropped.
[forwarded] = read(type="response.web_search_call.searching", output_index=0)
assert (forwarded.provider, forwarded.kind) == ("openai", "response.web_search_call.searching")

# The API sends no event meaning "the turn is over". The reader adds one up from the terminal
# event, and refuses a stream that ended without one.
read(type="response.completed", response={"status": "completed", "usage": {"output_tokens": 3}})
assert reader.finished().stop_reason.value == "end_turn"

Building the request

axio_responses.convert_messages(messages: list[Message], system: str) → tuple[str, list[dict[str, Any]]]

Convert axio Message list to Responses API input array.

Returns (instructions, input_items).

axio_responses.convert_tools(tools: list[Tool[Any]]) → list[dict[str, Any]]

Convert axio Tool list to Responses API function tool dicts.

axio.schema.strip_title(schema: dict[str, Any]) → dict[str, Any]

The schema without its title keywords, at every depth.

build_tool_schema writes none, so this is for a schema that came from somewhere else: a pydantic model names every model and field, no provider reads those names, and a large tool set pays for them on every request.

Only the keywords that hold schemas are walked. const, default, enum and examples hold values the caller declared, so a value with a title field of its own survives: walked as schemas, an enum of objects came out as a list of empty ones.

STOP_REASONS maps a published response status, or the reason an incomplete response gives, onto a StopReason:

A reason that is not in the map ends the turn as StopReason.error, never as end_turn. The event is named response.incomplete. The response did not finish. A truncation reason the API adds later must not be stored and reported as a whole answer.

Published	StopReason
completed, end_turn, stop	end_turn
max_output_tokens, length	max_tokens
cancelled	cancelled
content_filter	refusal

It is the only one of the four transport maps that names cancelled. A reason outside the map ends the run as an error rather than passing for a finished answer.

Reading the stream

class axio_responses.Responses

Every event the Responses API sends, and what each one becomes.

The vocabulary is the published ResponseStreamEvent union. An event missing from this body is one the API added after it was written, not one nobody named. A test reading with strict=True holds that against the schema.

One instance reads one response. Its state is the usage, the stop reason and the id map.

finished() → IterationEnd

What the turn added up to. The API sends no event that means this.

A stream that ended without one of its terminal events did not finish. The connection was cut. Reported as end_turn, a truncated answer is stored and returned as a whole one. So it is raised, the way a transport reports any other broken stream.

unmatched(name: str, payload: Payload) → Iterator[StreamEvent]

Anything this reader does not interpret, passed on under the provider’s own name.

Almost all of it is the API running a tool on its own side. Each such tool has its own event family. That set depends on which tools exist and which the caller declared, not on the protocol. Named one by one, the list goes stale the day a tool is added. It also reports a new tool as news about the protocol when it is news about the tools.

So nothing is listed and nothing is dropped. A consumer that wants the shell commands the model ran, or the searches it made, matches on kind. Any other consumer ignores it.

The reader names only what it interprets. Everything else — almost all of it the API running a tool on its own side — is forwarded by unmatched() as ProviderEvent(provider="openai", kind=<the API's own name>). See axio-sse for why.

Payload shapes

One axio_sse.Wire per payload, read field by declared name and type. A shape with no wire name of its own is only ever nested inside another.

Shape	Wire name
TextDeltaEvent	response.output_text.delta
ReasoningDeltaEvent	response.reasoning_summary_text.delta, response.reasoning_text.delta
RefusalDeltaEvent	response.refusal.delta
AnnotationAdded	response.output_text.annotation.added
ItemAdded	response.output_item.added
ItemDone	response.output_item.done
ContentPartDone	response.content_part.done
ArgumentsDelta	response.function_call_arguments.delta
ArgumentsDone	response.function_call_arguments.done
Created	response.created
Completed	response.completed
Incomplete	response.incomplete
Failed	response.failed
StreamFailure	error
ResponseObject	nested
ResponseUsage	nested
InputDetails, OutputDetails	nested
OutputItem	nested
Annotation, AnnotationSource	nested
IncompleteDetails, ResponseError	nested

class axio_responses.ResponseUsage(input_tokens: int = 0, output_tokens: int = 0, input_tokens_details: InputDetails = <factory>, output_tokens_details: OutputDetails = <factory>)

Both slices arrive inside their totals here, so the reader adds nothing to either.

class axio_responses.OutputItem(type: 'str' = '', id: 'str' = '', call_id: 'str' = '', name: 'str' = '', encrypted_content: 'str' = '', raw: 'Payload' = <factory>)

encrypted_content: str

Present on a reasoning item when the request asked for it. Opaque, and sent back on the next request.

raw: Payload

The item exactly as it arrived. An item this reader does not interpret has to go back whole on the next request, and no shape declared here would hold a type nobody has published yet.

class axio_responses.ResponseObject(id: 'str' = '', model: 'str' = '', status: 'str' = '', usage: 'ResponseUsage' = <factory>, output: 'list[OutputItem]' = <factory>, error: 'ResponseError' = <factory>, incomplete_details: 'IncompleteDetails' = <factory>)

class axio_responses.Annotation(text: str = '', title: str | None = None, url: str | None = None, file_id: str | None = None, source: AnnotationSource = <factory>, start_index: int | None = None, end_index: int | None = None, raw: Payload = <factory>)

One attribution. It arrives under a url shape, a file shape and the older flat citation, so each field is declared wherever a shape put it. The whole object travels in raw.

class axio_responses.AnnotationSource(url: 'str' = '', filename: 'str' = '')

class axio_responses.InputDetails(cached_tokens: 'int' = 0, cache_write_tokens: 'int' = 0)

class axio_responses.OutputDetails(reasoning_tokens: 'int' = 0)

class axio_responses.IncompleteDetails(reason: 'str' = '')

class axio_responses.ResponseError(message: 'str' = '', code: 'str' = '')

class axio_responses.TextDeltaEvent(delta: 'str' = '', output_index: 'int' = 0)

names: ClassVar[tuple[str, ...]] = ('response.output_text.delta',)

Every name this shape arrives under, from name= and also= on the class line.

output_index: int

Which output item this belongs to. Fixed at zero, every delta of a multi-item response shared one index while the events that close a block kept the real one.

class axio_responses.ReasoningDeltaEvent(delta: str = '', output_index: int = 0)

Both reasoning channels read the same. A model that sends the text rather than the summary would otherwise think in silence.

names: ClassVar[tuple[str, ...]] = ('response.reasoning_summary_text.delta', 'response.reasoning_text.delta')

Every name this shape arrives under, from name= and also= on the class line.

class axio_responses.RefusalDeltaEvent(delta: 'str' = '', output_index: 'int' = 0, raw: 'Payload' = <factory>)

names: ClassVar[tuple[str, ...]] = ('response.refusal.delta',)

Every name this shape arrives under, from name= and also= on the class line.

class axio_responses.AnnotationAdded(output_index: 'int' = 0, content_index: 'int' = 0, annotation: 'Annotation' = <factory>)

content_index: int

Which content part inside that item, which axio has no index of its own for.

names: ClassVar[tuple[str, ...]] = ('response.output_text.annotation.added',)

Every name this shape arrives under, from name= and also= on the class line.

output_index: int

Which output item the cited text belongs to. The deltas and the closing event are indexed by this, not by content_index.

class axio_responses.ItemAdded(output_index: 'int' = 0, item: 'OutputItem' = <factory>)

names: ClassVar[tuple[str, ...]] = ('response.output_item.added',)

Every name this shape arrives under, from name= and also= on the class line.

class axio_responses.ContentPartDone(output_index: 'int' = 0, content_index: 'int' = 0)

names: ClassVar[tuple[str, ...]] = ('response.content_part.done',)

Every name this shape arrives under, from name= and also= on the class line.

class axio_responses.ArgumentsDelta(item_id: 'str' = '', output_index: 'int' = 0, delta: 'str' = '')

names: ClassVar[tuple[str, ...]] = ('response.function_call_arguments.delta',)

Every name this shape arrives under, from name= and also= on the class line.

class axio_responses.ArgumentsDone(item_id: 'str' = '', name: 'str' = '', arguments: 'str' = '')

names: ClassVar[tuple[str, ...]] = ('response.function_call_arguments.done',)

Every name this shape arrives under, from name= and also= on the class line.

class axio_responses.Created(response: 'ResponseObject' = <factory>)

names: ClassVar[tuple[str, ...]] = ('response.created',)

Every name this shape arrives under, from name= and also= on the class line.

class axio_responses.Completed(response: 'ResponseObject' = <factory>)

names: ClassVar[tuple[str, ...]] = ('response.completed',)

Every name this shape arrives under, from name= and also= on the class line.

class axio_responses.Incomplete(response: 'ResponseObject' = <factory>)

names: ClassVar[tuple[str, ...]] = ('response.incomplete',)

Every name this shape arrives under, from name= and also= on the class line.

class axio_responses.Failed(response: 'ResponseObject' = <factory>)

names: ClassVar[tuple[str, ...]] = ('response.failed',)

Every name this shape arrives under, from name= and also= on the class line.

class axio_responses.StreamFailure(message: str = '', code: str = '')

The stream’s own error, which carries no response object.

names: ClassVar[tuple[str, ...]] = ('error',)

Every name this shape arrives under, from name= and also= on the class line.