keel-llm-adapter-openai

A reference keel-llm-protocol adapter for any OpenAI-compatible chat endpoint — OpenAI, Groq, OpenRouter, Mistral, Together, Fireworks, vLLM, Ollama (OpenAI mode), and more.

This package grounds the protocol standard in a working implementation. The status-code → error-taxonomy mapping, response/usage/finish-reason parsing, SSE streaming, and tool-call parsing here are real and tested against the actual OpenAI-compatible wire format — so keel-llm-protocol is proven against a real provider API, not asserted in the abstract.

Is this for you?

Adopt when — you're consuming via keel-llm-protocol / keel-llm-reliability and want one adapter that maps OpenAI / Groq / OpenRouter / Mistral / Together / Fireworks / vLLM / Ollama failures to the same typed taxonomy. Skip when — you're calling one provider directly in a simple app — the official openai SDK has a wider surface (batch, fine-tune, files, embeddings…); this adapter covers chat / streaming / tools.

Install

pip install keel-llm-adapter-openai     # pulls in keel-llm-protocol + httpx

Use

from keel_llm_adapter_openai import OpenAIAdapter
from keel_llm_protocol import user
from keel_llm_protocol.errors import RateLimitError, AdapterError

# Point base_url at any OpenAI-compatible endpoint:
adapter = OpenAIAdapter(
    model="llama-3.3-70b-versatile",
    api_key="...",
    base_url="https://api.groq.com/openai/v1",
    provider="groq",                       # -> model_key = "groq:llama-3.3-70b-versatile"
)

try:
    resp = await adapter.generate([user("Explain circuit breakers in one sentence.")])
    print(resp.text, resp.usage.total_tokens, resp.finish_reason)
except RateLimitError as e:
    ...                                    # healthy but throttled — back off; don't trip a breaker
except AdapterError as e:
    if e.retryable:
        ...                                # transient / timeout — retry or fail over

It implements all three capability protocols, so it type-checks anywhere any of them is required:

from keel_llm_protocol import ModelAdapter, StreamingModelAdapter, ToolCallingModelAdapter
assert isinstance(adapter, ModelAdapter)
assert isinstance(adapter, StreamingModelAdapter)
assert isinstance(adapter, ToolCallingModelAdapter)

Streaming

async for chunk in adapter.stream([user("Write a haiku.")]):
    print(chunk.delta, end="", flush=True)
    if chunk.usage:
        print(f"\n[{chunk.usage.total_tokens} tokens]")

Tools

from keel_llm_protocol import ToolSpec
weather = ToolSpec(name="get_weather", description="Get weather",
                   parameters={"type": "object", "properties": {"city": {"type": "string"}}})
resp = await adapter.generate_with_tools([user("Weather in SF?")], [weather])
for call in resp.tool_calls:
    print(call.name, call.arguments)      # YOU dispatch — the adapter only reports the request

Error mapping

Provider HTTP responses are mapped to the standard taxonomy so reliability logic is provider-agnostic:

A 200 response whose finish_reason is content_filter is returned (not raised) with finish_reason="content_filter".

Status

0.1.2 — 0.x while the API stabilizes through year one (breaking changes possible at minor bumps, documented in the CHANGELOG; pin exact versions).

The Keel toolkit

Composable, vendor-neutral LLM reliability libraries on PyPI: keel-llm-reliability · keel-llm-protocol · keel-llm-adapter-openai · keel-llm-adapter-anthropic · keel-llm-adapter-google · keel-circuit-breaker

MIT licensed.