axio-transport-openai

OpenAI and OpenAI-compatible completion transports.

api selects the endpoint. OpenAITransport defaults to "responses" and posts to /v1/responses, reading it through axio-responses. NebiusTransport, OpenRouterTransport and OpenAICompatibleTransport set "chat" and post to /v1/chat/completions. That endpoint refuses function tools beside any reasoning effort other than "none" — see the troubleshooting entry.

extra_params is folded into the request. Its tools are merged with the agent’s function declarations rather than replacing them, so adding a hosted tool does not take away the calls the agent is there to dispatch.

class axio_transport_openai.OpenAITransport(name: 'str' = 'OpenAI', base_url: 'str' = <factory>, api_key: 'str' = <factory>, model: 'ModelSpec' = <factory>, models: 'ModelRegistry' = <factory>, session: 'aiohttp.ClientSession | None' = None, max_retries: 'int' = 10, retry_base_delay: 'float' = 5.0, extra_params: 'Mapping[str, Any]'=mappingproxy({}), *, api: "Literal['responses', 'chat'] | None"=None, reasoning_summary: "Literal['auto', 'concise', 'detailed']"='auto')

api: Literal['responses', 'chat'] | None

Which endpoint this server speaks. "responses" is the one that takes function tools and reasoning together, and OpenAI recommends it for new work. Left unset it follows base_url: only OpenAI’s own host is assumed to implement it.

build_responses_payload(messages: list[Message], tools: list[Tool[Any]], system: str) → dict[str, Any]

The request /v1/responses takes.

The system prompt goes in instructions rather than in a message. Tool calls and their outputs are items beside the messages rather than blocks inside them. This endpoint takes tools and reasoning together, which is the reason to prefer it. /v1/chat/completions refuses that pair outright for a model that reasons.

async embed(texts: list[str]) → list[list[float]]

Call the OpenAI-compatible /v1/embeddings endpoint.

reasoning_summary: Literal['auto', 'concise', 'detailed']

How much of its reasoning the model summarises for a reader on /v1/responses. The raw chain is never returned; without a summary there is nothing to show.

class axio_transport_openai.nebius.NebiusTransport(name: 'str' = 'Nebius AI Studio', base_url: 'str' = 'https://api.tokenfactory.nebius.com/v1', api_key: 'str' = <factory>, model: 'ModelSpec' = <factory>, models: 'ModelRegistry' = <factory>, session: 'aiohttp.ClientSession | None' = None, max_retries: 'int' = 10, retry_base_delay: 'float' = 5.0, extra_params: 'Mapping[str, Any]'=mappingproxy({}), thinking: 'bool' = False, *, api: "Literal['responses', 'chat']"='chat', reasoning_summary: "Literal['auto', 'concise', 'detailed']"='auto')

async fetch_models() → None

Fetch available models from Nebius /v1/models?verbose=true.

class axio_transport_openai.openrouter.OpenRouterTransport(name: 'str' = 'OpenRouter', base_url: 'str' = 'https://openrouter.ai/api/v1', api_key: 'str' = <factory>, model: 'ModelSpec' = ModelSpec(id='google/gemini-2.5-pro-preview', capabilities=frozenset(), max_output_tokens=8192, context_window=128000, input_cost=0.0, output_cost=0.0), models: 'ModelRegistry' = <factory>, session: 'aiohttp.ClientSession | None' = None, max_retries: 'int' = 10, retry_base_delay: 'float' = 5.0, extra_params: 'Mapping[str, Any]'=mappingproxy({}), thinking: 'bool' = False, *, api: "Literal['responses', 'chat']"='chat', reasoning_summary: "Literal['auto', 'concise', 'detailed']"='auto')

async fetch_models() → None

Fetch available models from OpenRouter /v1/models.

resolve_model(model_id: str) → ModelSpec

Resolve OpenRouter model IDs, including routed variants such as :nitro.

class axio_transport_openai.custom.OpenAICompatibleTransport(name: str = 'OpenAI', base_url: str = '', api_key: str = <factory>, model: ModelSpec = <factory>, models: ModelRegistry = <factory>, session: aiohttp.ClientSession | None = None, max_retries: int = 10, retry_base_delay: float = 5.0, extra_params: Mapping[str, Any]=mappingproxy({}), *, api: Literal['responses', 'chat']='chat', reasoning_summary: Literal['auto', 'concise', 'detailed']='auto')

OpenAI-compatible transport for a single user-defined provider.

Instances are created by CustomHubScreen with name, base_url, api_key, and models populated from the JSON config. Supports JSON round-trip via to_dict() / from_dict().

api: Literal['responses', 'chat']

Which endpoint this server speaks. "responses" is the one that takes function tools and reasoning together, and OpenAI recommends it for new work. Left unset it follows base_url: only OpenAI’s own host is assumed to implement it.

classmethod from_dict(data: dict[str, Any], *, session: ClientSession | None = None) → Self

As OpenAITransport.from_dict(), but base_url/api_key come verbatim from data.

The base implementation reads a value saved empty as empty, and falls back to the OPENAI_BASE_URL/OPENAI_API_KEY env vars only where the key is absent altogether. That is right for the built-in OpenAI provider, whose settings dict is written by hand and omits what it wants the default for.

A custom provider’s config is a full round-trip of to_dict(), so an absent key means the same thing as an empty one: this server takes no credential. Reading it through the environment would point a local server at an unrelated real endpoint.

reasoning_summary: Literal['auto', 'concise', 'detailed']

How much of its reasoning the model summarises for a reader on /v1/responses. The raw chain is never returned; without a summary there is nothing to show.

ChatGPT subscription

ChatGPTTransport uses a pinned ChatGPTSession. Ordinary sign-in uses the public Responses endpoint; axio-chatgpt login --codex uses native ChatGPT OAuth and the Codex backend for completion and image generation. It discovers visible models for that account, builds subscription-compatible requests, and delegates HTTP/SSE recovery to ChatGPTResponses. Model copies share the account session and catalog; account changes create a separate catalog. Subscription prices are unknown (NaN). All server completion models receive function tools. Vision, reasoning, and context limits come from the API metadata instead of the API-key registry.

Install axio-transport-openai and run axio-chatgpt login to save and select a registration in the default XDG credentials file. Use --manual for sign-in over SSH. The command prints an issued client ID after successful sign-in. Keep this ID as the account reference; keep tokens in the credential store.

Use axio-chatgpt accounts, select CLIENT_ID, status, models, and logout to manage saved accounts. status is offline; models fetches the account catalog. These commands accept --credentials PATH. Status and model listing accept --json and --account CLIENT_ID; the latter does not change the saved selection. Reauthorize a registration with axio-chatgpt login --account CLIENT_ID.

import aiohttp
from axio import Agent, MemoryContextStore
from axio_transport_openai import ChatGPTTransport
from axio_transport_openai.credential_stores import FileCredentialStore
from axio_transport_openai.oauth import ChatGPTOAuth
from axio_transport_openai.session import ChatGPTSession


async def ask_saved_account(client_id: str, model_id: str) -> str:
    store = FileCredentialStore()
    async with aiohttp.ClientSession() as http:
        oauth = ChatGPTOAuth(http, await store.host_id())
        async with ChatGPTSession(store, oauth, client_id) as account:
            transport = ChatGPTTransport(account=account, session=http)
            await transport.fetch_models()
            selected = transport.with_model(model_id)
            agent = Agent(system="Be concise.", transport=selected)
            return await agent.run("Hello", MemoryContextStore())

Choose model_id from that account’s visible catalog. Image input is enabled by the catalog’s input_modalities. Explicit model_specs supply fallback metadata for missing fields; a new model ID does not disable tools. Session closure stops background renewal and retains credentials for the next process. Conversation persistence is separate; use a persistent context store or serialize Message objects to resume history.

PlanConsentRequired means identity sign-in succeeded without subscription permission. SignInRequired requires reauthorization of the selected registration. ResponsesAPIError preserves the HTTP status, provider code, and request ID. Quota exhaustion pauses inference for the shared account session. Call resume_plan_usage() only after resolving the limit. Errors never select another account or switch to API-key billing.

Run axio-repl --transport chatgpt --model gpt-6-luna to use the selected account in the CLI. Scripts can use accounts.open_chatgpt(store, http, model=...) to resolve the selected registration, fetch its catalog, and manage session lifetime. It also accepts client_id= for an explicit account. Selection changes affect future sessions only. TUI Configure ChatGPT provides sign-in, registration selection, permission status and sign-out. The registry opens managed account sessions and calls with_model() for model copies; generic config stores only account references. The native sign-in mode saves credentials in the same Axio store. It does not import the separate axio-transport-codex transport’s credentials.

Images

Authorize with axio-chatgpt login --codex before generating images. Native login uses port 1455 by default; --manual supports SSH. The ordinary public plan-sharing token is not accepted by the native image endpoint. The account’s authorization mode is persisted and determines routing after restart.

ChatGPTImageTransport implements ImageGenTransport. Its models registry is CHATGPT_IMAGE_MODELS, currently containing gpt-image-2 as a ChatGPTImageModel with Capability.image_generation and Capability.image_editing. Native ChatGPTTransport.fetch_models() also appends these image presets to the completion catalog. They describe implemented image support, not guaranteed account entitlement. Filter completion choices by Capability.text and image choices by Capability.image_generation.

import aiohttp
from axio_transport_openai import ChatGPTImageTransport
from axio_transport_openai.chatgpt_images import image_media_type
from axio_transport_openai.session import ChatGPTSession


async def draw(account: ChatGPTSession, http: aiohttp.ClientSession) -> tuple[bytes, str]:
    generator = ChatGPTImageTransport(account=account, session=http)
    images = await generator.generate_images("An owl reading a book", quality="low")
    return images[0], image_media_type(images[0])

For an open completion transport, chatgpt.image_transport() returns this class with the same sessions. chatgpt.with_model("gpt-image-2").generate_images(prompt) also returns image bytes. Image models reject stream() calls.

One image per request (n=1) is supported. The standalone generator accepts size (auto, 1024x1024, 1536x1024, 1024x1536), quality (auto, low, medium, high), and transparent_background. Native output is PNG. Output format selection, partial images, and seeds are not exposed.

The protocol follows Codex CLI rust-v0.163.0 and posts to /backend-api/codex/images/generations on ChatGPT. Backend access and quotas depend on the account. ChatGPTImageError preserves API diagnostics; invalid results raise an error. Quota errors pause the shared account session. Generation POSTs are never automatically replayed, including after server errors or cancellation. A connection failure can leave the server outcome unknown. The caller owns both sessions, and closing them retains credentials.

ChatGPTImageTransport also implements ImageEditTransport from axio.transport. Use edit_images(prompt, images=[ImageBlock(...)]) for changes to existing images. Sources are required and retain their order; PNG/JPEG/WebP bytes are sent as inline image references to /backend-api/codex/images/edits. ChatGPTTransport.edit_images() delegates to the same implementation. The standalone class accepts the generation size/quality/background settings. The native protocol has no mask field; mask must remain None. Editing shares the account’s quota gate and never retries requests automatically. image_editing does not imply vision.

ChatGPTImageTools(chatgpt.image_transport()) creates two Tool instances in .tools: generate_image(prompt) and edit_image(prompt, images). Attach these to Agent.tools. The model supplies image IDs to the edit tool; the host registers the actual attachments with add_image(ImageBlock(...)) and includes their IDs in the conversation. Results contain new IDs and ImageBlocks for subsequent edits. Use one tool set per conversation. Its read-only .images catalog can be persisted by the host and restored with the constructor’s images= argument. No filesystem paths, URLs, or raw base64 are requested from the model. Both tools declare external access, so standard tool guards apply.

class axio_transport_openai.ChatGPTTransport(*, account: ChatGPTSession, session: ClientSession, model: str | None = None, model_specs: Iterable[ModelSpec] = (), base_url: str | None = None, image_base_url: str = 'https://chatgpt.com/backend-api/codex', max_attempts: int = 3, retry_base_delay: float = 2.0, reasoning_effort: str | None = None, reasoning_summary: Literal['auto', 'concise', 'detailed'] = 'auto')

A completion transport pinned to one authenticated account session.

The caller owns the HTTP and account sessions. with_model() shares both and the account catalog. with_account() starts a new catalog without changing existing transports. No API-key, embedding, or realtime interface is exposed.

build_payload(messages: list[Message], tools: list[Tool[Any]], system: str) → dict[str, Any]

Build a request after fetching and selecting a visible account model.

property catalog: tuple[ChatGPTModel, ...]

Server entries and native image presets, with visibility and display names.

async edit_images(prompt: str, *, images: list[ImageBlock], mask: ImageBlock | None = None, model: str | None = None, n: int = 1) → list[bytes]

Edit supplied images with the selected image model, or GPT Image 2.

async fetch_models() → None

Refresh this account’s catalog. Failure invalidates it and propagates.

static format_error(error: Exception) → str

Return safe account recovery text for interactive applications.

async generate_images(prompt: str, *, model: str | None = None, n: int = 1) → list[bytes]

Generate with the selected image model, or the default GPT Image model.

image_transport(model: str | ModelSpec = 'gpt-image-2') → ChatGPTImageTransport

Create a standalone generator sharing this account and HTTP session.

property models: ModelRegistry

Visible server models followed by native image presets; empty until fetched.

classmethod open_for_app(*, session: ClientSession, client_id: str | None = None, account: str | None = None, credentials: str | None = None) → AbstractAsyncContextManager[ChatGPTTransport]

Open an account session for an application that owns transport lifetimes.

static select_model(transport: ChatGPTTransport, model: ModelSpec) → ChatGPTTransport

Reset an incompatible reasoning setting when an application changes models.

static supported_thinking_levels(model: ModelSpec | None) → tuple[str, ...]

Use catalog levels when present; otherwise retain the generic fallback.

with_account(account: ChatGPTSession, *, model: str | None = None) → ChatGPTTransport

Create a separately fetched catalog for another selected account.

with_model(model: str | ModelSpec) → ChatGPTTransport

Select a visible catalog model while sharing the current account session.

class axio_transport_openai.chatgpt_models.ChatGPTModel(id: str, capabilities: frozenset[Capability] = frozenset({}), max_output_tokens: int = 8192, context_window: int = 128000, input_cost: float = nan, output_cost: float = nan, display_name: str = '', visibility: str = '', metadata: Mapping[str, ~typing.Any]=<factory>)

A server catalog entry with capabilities and limits from API metadata.

Subscription costs are unknown, represented as NaN rather than zero. Generic ModelSpec token budgets remain fallbacks when no model metadata is available. All catalog models support text and function tools. Missing metadata can use an explicitly supplied ModelSpec. The API-key registry is not consulted.

class axio_transport_openai.ChatGPTImageModel(id: str, capabilities: frozenset[Capability] = frozenset({Capability.image_editing, Capability.image_generation}), max_output_tokens: int = 0, context_window: int = 0, input_cost: float = nan, output_cost: float = nan, display_name: str = '', visibility: str = 'list', metadata: Mapping[str, ~typing.Any]=<factory>)

A native ChatGPT Images model, separate from completion catalog metadata.

Token budgets and token prices do not describe image generation. Availability is checked by the image endpoint for the selected account on each request.

class axio_transport_openai.ChatGPTImageTransport(*, account: ChatGPTSession, session: ClientSession, model: str | ModelSpec = 'gpt-image-2', base_url: str = 'https://chatgpt.com/backend-api/codex')

Generate image bytes using native ChatGPT OAuth, with no API-key fallback.

The caller owns both sessions. Models are supported image endpoint presets, not a promise of account quota. POST requests are never automatically replayed: a lost response, cancellation, or server error may hide a completed generation.

async edit_images(prompt: str, *, images: list[ImageBlock], mask: ImageBlock | None = None, model: str | None = None, n: int = 1, size: Literal['auto', '1024x1024', '1536x1024', '1024x1536'] = 'auto', quality: Literal['auto', 'low', 'medium', 'high'] = 'auto', transparent_background: bool = False) → list[bytes]

Edit one or more PNG/JPEG/WebP sources using native ChatGPT Images.

The native Codex protocol does not expose masks. Passing one fails before authorization or HTTP; sources are never silently treated as a text prompt.

async generate_images(prompt: str, *, model: str | None = None, n: int = 1, size: Literal['auto', '1024x1024', '1536x1024', '1024x1536'] = 'auto', quality: Literal['auto', 'low', 'medium', 'high'] = 'auto', transparent_background: bool = False) → list[bytes]

Return complete images; use image_media_type() to identify their format.

with_model(model: str | ModelSpec) → ChatGPTImageTransport

Select an image model while sharing account renewal and quota state.

class axio_transport_openai.ChatGPTImageTools(transport: ChatGPTImageTransport, *, images: Mapping[str, ImageBlock] | None = None)

Bind two model-facing tools to an image transport and a local image catalog.

The host registers attachments with add_image() and tells the agent their IDs. Tool results contain new image IDs and ImageBlocks, allowing subsequent edits without model-generated base64 or filesystem access. Create one instance per conversation. The host owns session lifetime and persists images if desired.

add_image(image: ImageBlock, *, image_id: str | None = None) → str

Register an attachment and return its ID for the agent’s prompt.

async edit_image(prompt: ~typing.Annotated[str, ~axio.field.FieldInfo(description=, default=MISSING, ge=None, le=None, strict=True)], images: list[str]) → list[TextBlock | ImageBlock]

Edit existing images according to the requested changes. images must contain IDs registered by the application or returned by earlier image tool calls. Keep their order when referring to first/second images. Returns a new image and ID; source images remain available. Masks are unsupported.

async generate_image(prompt: ~typing.Annotated[str, ~axio.field.FieldInfo(description=, default=MISSING, ge=None, le=None, strict=True)]) → list[TextBlock | ImageBlock]

Create a new image from a description when the user requests generation. To change an existing image or use reference images, call edit_image instead. Returns an image and its ID, which can be used in later edits.

property images: Mapping[str, ImageBlock]

Read-only catalog of attachments and generated results, keyed by ID.

axio_transport_openai.chatgpt_images.image_media_type(data: bytes) → ImageMediaType

Identify a PNG, JPEG, or WebP result without relying on a file extension.

axio_transport_openai.chatgpt.build_chatgpt_payload(messages: list[Message], tools: list[Tool[Any]], system: str, model: ModelSpec, *, reasoning_effort: str | None = None, reasoning_summary: str = 'auto') → dict[str, Any]

Declare current tools at input position zero, then replay complete history.

Keeping this position stable makes unchanged tool sets reusable across turns. Selector or schema changes replace this declaration in place; old calls remain in history, but removed tools are no longer declared. No tool-loading state is held in the transport, so forks, restarts, and concurrent contexts are isolated.

class axio_transport_openai.session.ChatGPTSession(store: CredentialStore[CredentialRecord], oauth: ChatGPTOAuth, client_id: str, *, refresh_margin: float = 60, retry_delay: float = 5)

Keep one issued client registration current for the lifetime of a run.

Use as an async context manager to enable background renewal. Without it, access_token() still renews on demand. close() stops background work without signing out. The caller owns the OAuth client’s aiohttp session and must keep it open until close() finishes. last_error reports background failures.

async access_token() → str

Get a current plan token, reloading credentials under the process lock.

property client_id: str

The registration pinned when this session was created.

async close() → None

Stop renewal and finish any credential commit already in progress.

async request_headers(*, images: bool = False) → dict[str, str]

Resolve token and workspace together from the pinned registration.

resume_plan_usage() → None

Allow another plan request after the caller resolves a reported usage limit.

A quota response provides no reliable reset time. Resuming is explicit; it does not change accounts, credentials, or the billing path.

async sign_out() → SignOutResult

Revoke the latest renewable session and retain its registration.

Temporary revocation errors get two retries. Local tokens are cleared even if the server cannot confirm revocation. A storage failure raises CredentialError; it never reports successful local sign-out.

async start() → None

Start background renewal after checking the selected registration.

property supports_images: bool

Whether the last validated credentials use native ChatGPT authorization.

class axio_transport_openai.credential_stores.FileCredentialStore(path: str | Path | None = None, *, record_type: type[T] | None = None)

class axio_transport_openai.credential_stores.SQLiteCredentialStore(path: str | Path, *, record_type: type[T] | None = None)

Realtime

class axio_transport_openai.OpenAIRealtimeTransport(api_key: str = <factory>, base_url: str = <factory>, model: str = 'gpt-realtime', http_session: ClientSession | None = None, server_vad: bool = True, transcription_model: str | None = None, vad_threshold: float = 0.5, vad_prefix_padding_ms: int = 300, vad_silence_duration_ms: int = 500, vad_create_response: bool = True, vad_interrupt_response: bool = True, input_noise_reduction: str | None = 'near_field')

RealtimeTransport for OpenAI’s Realtime WebSocket API.

Server-VAD knobs live here so callers can tune barge-in sensitivity without touching the transport internals. When AEC is imperfect or the room is loud, the defaults can cause the model’s own audio to trip interrupt_response and cancel its response. Bump vad_threshold higher, lengthen vad_silence_duration_ms, or set vad_interrupt_response=False to require an explicit agent.interrupt().

input_noise_reduction: str | None

Server-side noise reduction profile applied to mic audio. Options: "near_field" (close-talking mic / headset), "far_field" (laptop or conference mic), or None to disable.

server_vad: bool

Set False for explicit utterance boundaries via commit().

transcription_model: str | None

Opt in to user transcripts, e.g. gpt-4o-mini-transcribe.

vad_create_response: bool

Whether the server auto-creates a response after each user turn.

vad_interrupt_response: bool

Whether incoming user speech cancels the model’s in-flight response. Setting this to False makes interruption explicit (call agent.interrupt()) — useful when AEC isn’t strong enough and the model’s own voice would otherwise loop back through the mic and interrupt itself.

vad_prefix_padding_ms: int

Audio retained before a detected speech start.

vad_silence_duration_ms: int

Silence required before the server considers a turn complete.

vad_threshold: float

Server VAD activation threshold (0..1). Higher → less sensitive to quiet sounds and bleed-through. Default 0.5.

class axio_transport_openai.OpenAIRealtimeSession(ws: Any, input_audio_media_type: str = 'audio/pcm;rate=24000', output_audio_media_type: str = 'audio/pcm;rate=24000', http_session: ClientSession | None = None, own_http_session: bool = False)

RealtimeSession backed by an aiohttp WebSocket.

ws may be any object that quacks like aiohttp.ClientWebSocketResponse (send_str, async iteration yielding aiohttp.WSMessage-like objects, close, closed). Tests inject a stub.

async close() → None

Tear down the session and release resources.

async commit() → None

Signal end-of-utterance for manual VAD; no-op with server VAD.

async events() → AsyncIterator[StreamEvent]

Async iterator over server events for the lifetime of this session.

async interrupt() → None

Abort in-flight assistant generation (e.g. user interrupted).

async send(content: ContentBlock | list[ContentBlock]) → None

Append user content (audio chunk, text, image) to the input buffer.

async send_tool_result(tool_use_id: ToolCallID, name: ToolName, content: str | list[ContentBlock]) → None

Deliver a tool’s result to the provider so generation can resume.

name is included because some providers (e.g. Gemini Live) require the tool name alongside the call id. OpenAI realtime can ignore it.

Standalone API speech

These API-key transports are independent of completion and ChatGPT subscription credentials. See the package speech examples for billing boundaries, encoding and streaming cleanup requirements.

class axio_transport_openai.OpenAISTTTransport(*, api_key: str = <factory>, base_url: str = <factory>, session: ClientSession | None = None, timeout_seconds: float = 120.0, model: str = 'gpt-4o-mini-transcribe', language: str | None = None, prompt: str | None = None)

Transcribe an encoded audio file via /audio/transcriptions (not raw PCM).

class axio_transport_openai.OpenAITTSTransport(*, api_key: str = <factory>, base_url: str = <factory>, session: ClientSession | None = None, timeout_seconds: float = 120.0, model: str = 'gpt-4o-mini-tts', voice: str = 'coral', response_format: Literal['mp3', 'opus', 'aac', 'flac', 'wav', 'pcm']='mp3', speed: float = 1.0, instructions: str | None = None)

Stream binary audio via /audio/speech. No replay/retry of paid requests.

PCM output is signed 16-bit little-endian mono at 24 kHz without a header. Network chunks are arbitrary byte boundaries, not PCM frames or standalone files. Close an abandoned generator with contextlib.aclosing to release its response.