🪄 callm

A small, type-safe Python interface to the chat models, built on pydantic-ai.

callm is the contract, not the transport. Five providers reach you through one ChatModel: called for a turn, streamed for incremental results, and the same either way. The wire protocols underneath are pydantic-ai's, which is why there is so little here to go wrong.

Features:

• One interface across OpenAI, Codex and Azure OpenAI
• Type-safe structured output with Pydantic
• Tool calling
• Async streaming, ending in the same response a call returns
• Images, reasoning traces and cache-aware token usage
• Automatic retries for transient failures, with per-retry callbacks

Contents

• Installation
• Quick start
• The contract
  • Messages
  • Calling
  • Streaming
  • Structured output
  • Tools
  • Images
  • Usage and reasoning
  • Errors and retries
• Providers
• Model settings
• Credits
• License

Installation

pip install pycallm

This covers OpenAI, Azure OpenAI and Codex. The Codex WebSocket transport needs one more package:

pip install pycallm[websocket]

Quick start

import asyncio
from callm import ChatOpenAI, SystemMessage, UserMessage

async def main():
    async with ChatOpenAI("gpt-6-sol") as model:
        response = await model.call([
            SystemMessage(content="You are a helpful assistant."),
            UserMessage(content="What is 2+2?"),
        ])

    print(response.completion)          # "2 + 2 equals 4."
    print(response.usage.total_tokens)  # 29

asyncio.run(main())

The contract

Everything public lives in callm.messages and callm.base, and every provider speaks exactly it.

Messages

Four message types, all frozen, so a history can be shared between requests without one of them editing another's:

from callm import SystemMessage, UserMessage, AssistantMessage, ToolResultMessage

messages = [
    SystemMessage(content="You are a Python expert."),
    UserMessage(content="How do I read a file?"),
    AssistantMessage(content="Use open() with a context manager."),
    UserMessage(content="Show me an example."),
]

A SystemMessage is carried as the request's instructions rather than as a turn in the history, which is what every provider actually wants.

Calling

call runs one turn and returns it whole:

response = await model.call(messages)

A ModelResponse carries:

Field	What it is
completion	the text, or the parsed object when output_format was given (None while a turn with tools is still calling them)
thinking	the reasoning trace, when the model exposed one
tool_calls	what the model wants run before it can finish
usage	input, output and cache token counts
finish_reason	why the model stopped
provider_state	the provider's own turn, for replaying it verbatim

response.as_assistant_message() turns a turn into the history entry for the next request, provider_state included — which is what keeps a reasoning model from losing its train of thought across a tool round-trip.

Streaming

A stream yields deltas as they arrive and ends with exactly one ModelResponse, the same value call would have returned. Tool calls are never streamed half-built: each one arrives once its arguments are complete.

from callm import ModelEventType

async for event in model.stream(messages):
    match event.type:
        case ModelEventType.TEXT_DELTA:
            print(event.delta, end="", flush=True)
        case ModelEventType.THINKING_DELTA:
            ...  # reasoning, kept separate from the answer
        case ModelEventType.TOOL_CALL:
            print(event.tool_call.name)
        case ModelEventType.RESPONSE:
            print(event.usage.total_tokens)

Structured output

Hand call a Pydantic model and get one back:

from pydantic import BaseModel

class Recipe(BaseModel):
    name: str
    minutes: int
    ingredients: list[str]

response = await model.call(messages, output_format=Recipe)
response.completion.ingredients  # list[str]

Every provider does this through a tool call, since that is the one shape all of them speak. An answer that does not parse raises ModelBehaviorError rather than arriving as something it is not.

Tools

A tool is a name, a description and a JSON schema. Running the calls the model asks for is up to you:

from callm import ModelTool, ToolResultMessage

search_web = ModelTool(
    name="search_web",
    description="Search the web for information.",
    parameters={
        "type": "object",
        "properties": {"query": {"type": "string", "description": "What to look for"}},
        "required": ["query"],
    },
)

messages = [UserMessage(content="Look up callm and summarise it.")]

while True:
    response = await model.call(messages, tools=[search_web])
    messages.append(response.as_assistant_message())
    if not response.tool_calls:
        break
    for call in response.tool_calls:
        query = call.parsed_arguments["query"]
        messages.append(ToolResultMessage(
            tool_call_id=call.id,
            tool_name=call.name,
            content=f"results for {query}",
        ))

tool_choice takes "auto", "required" or "none".

Images

from callm import ImageUrl, UserMessage

UserMessage(content=(
    "What's in this image?",
    ImageUrl(url="https://example.com/photo.jpg", detail="high"),
))

A data: URI works the same way and is sent as bytes.

Usage and reasoning

Usage reports input_tokens, output_tokens, cache_read_tokens, cache_write_tokens and a total_tokens property. Every provider fills the same fields; a counter a provider does not report stays zero.

Errors and retries

Provider failures arrive as callm errors, whichever SDK raised them:

Error	Raised when
AuthenticationError	credentials rejected (401, 403)
CredentialsUnavailableError	credentials missing or unusable, a new login is needed
RateLimitError	temporary 429 — retryable
RetryableError	5xx, 408, 409, 425, transport failures
OutOfCreditsError	quota or billing exhausted
ContextLengthExceededError	the input did not fit
ModelBehaviorError	the answer did not fit the shape it was asked for
ProviderError	another nonretryable provider HTTP error
ResponseInterruptedError	a Codex WebSocket response started, then the connection broke; never replayed automatically

Retryable failures are retried with exponential backoff, honouring Retry-After when the provider sends one. A stream is only retried while nothing has been emitted yet, so output is never replayed. on_retry must be an async callable; it receives the delay in seconds and one-based attempt numbers. The SDK's own retries are disabled so the callback sees every retry.

async def log_retry(event):
    print(f"attempt {event.failed_attempt}/{event.max_attempts} failed, "
          f"retrying in {event.delay:.1f}s: {event.error.code}")

model = ChatOpenAI("gpt-6-sol", max_retries=3, on_retry=log_retry)

For UI messages, use event.error.user_message and event.delay. Use event.error.code to localize the message. Exception strings can include raw provider details and belong in private diagnostics rather than UI text.

Providers

Every constructor takes the model name first and falls back to the usual environment variable for credentials.

from callm import (
    ChatOpenAI,            # OPENAI_API_KEY
    ChatOpenAIResponses,   # OPENAI_API_KEY — the Responses API
    ChatAzureOpenAI,       # AZURE_OPENAI_API_KEY + AZURE_OPENAI_ENDPOINT
    ChatAzureOpenAIResponses,
    ChatCodex,             # a ChatGPT subscription
)

model = ChatOpenAI("gpt-6-sol", api_key="sk-...", base_url="https://...")
model = ChatAzureOpenAI("my-deployment", api_version="2024-10-01")

Reasoning models. ChatOpenAIResponses (and ChatCodex, and ChatAzureOpenAIResponses) take reasoning_effort — "none", "minimal", "low", "medium", "high", "xhigh" or "max" — and reasoning_summary. Which levels a model accepts differs, and an unsupported one comes back as a request error. Prefer the Responses API over ChatOpenAI for these models: it carries reasoning state between turns.

Codex. A reverse-engineered endpoint that authenticates with a ChatGPT subscription rather than an API key; OpenAI neither documents nor supports it. If the Codex CLI is logged in, its session is borrowed:

model = ChatCodex("gpt-6-sol", reasoning_effort="high")

ChatCodex uses HTTP by default. Set transport="websocket" to reuse a WebSocket connection across turns; this needs the websocket extra. If opening that connection or starting a response fails, it retries the request over HTTP. A connection lost after a response starts is reported without replaying the request.

For a conversation where the next user message has not arrived yet, prepare() can send the existing context over the WebSocket without generating an answer. The next matching turn reuses that prepared context:

from callm import ChatCodex, Message, SystemMessage, UserMessage

async with ChatCodex("gpt-6-sol", transport="websocket") as model:
    history: list[Message] = [SystemMessage(content="Answer briefly.")]
    await model.prepare(history)
    history.append(UserMessage(content="Name one European capital."))
    answer = await model.call(history)

Pass the same tools, tool_choice, and output_format to prepare() as to the next call when using them. If the history or connection changes, the next request sends the full history. With transport="http", prepare() is a no-op. See the WebSocket example for multiple turns. Pass an async on_transport_fallback callback to receive a TransportFallbackEvent with phase ("prepare", "call", or "stream") and reason. A failed preparation reports its error; call() and stream() report when they actually switch to HTTP. The example prints these events.

This reads ~/.codex/auth.json (honouring CODEX_HOME) but never writes it, so refreshed tokens last only as long as the process. To keep them, pass a credential_source — any OpenAICodexCredentialSource, as described in pydantic-ai's docs. A missing or unusable login raises CredentialsUnavailableError.

Model settings

Shared pydantic-ai settings are named, keyword-only parameters on every model constructor. The IDE can show their types and defaults:

model = ChatOpenAI(
    "gpt-6-sol",
    reasoning_effort="none",
    max_tokens=1000,
    temperature=0.7,
    stop_sequences=["\n\n"],
    timeout=60.0,
    max_retries=2,
    extra_headers={"X-Tenant": "acme"},
)

A setting's availability depends on the provider and model. Other provider-specific options are passed to pydantic-ai as model settings:

model = ChatOpenAIResponses("gpt-6-sol", openai_text_verbosity="low")

See pydantic-ai's model settings for the full list.

Credits

Built on pydantic-ai. Inspired by LangChain and browser-use.

License

MIT