Python, in your code

Run the gateway inside your Python process and call it with the OpenAI or Anthropic SDK. No server, no socket.

pip install "respan-gateway[openai]"
export ANTHROPIC_API_KEY=sk-ant-...

import respan_gateway as rg

with rg.Gateway() as gw:
    client = gw.openai()
    reply = client.chat.completions.create(
        model="claude-haiku-5-5",
        messages=[{"role": "user", "content": "hi"}],
    )
    print(reply.choices[0].message.content)

That is a Claude model called through the OpenAI SDK. The gateway translates the request to Anthropic’s Messages API and the reply back to Chat Completions. The request never leaves your process until the gateway calls Anthropic.

The package is the Rust gateway compiled into a native extension (abi3), for Python 3.10 and later. It also installs the respan-gateway command.

Install

pip install respan-gateway                 # the gateway and httpx
pip install "respan-gateway[openai]"       # plus the openai SDK, for gw.openai()
pip install "respan-gateway[anthropic]"    # plus the anthropic SDK, for gw.anthropic()

gw.openai() and gw.anthropic() import their SDK when you call them, so you only need the one you use.

Clients

Each method returns the official SDK client, wired to the gateway in-process. Use it as you would normally.

Any model works from any of them. A GPT model through the Anthropic SDK:

with rg.Gateway() as gw:
    client = gw.anthropic()
    message = client.messages.create(
        model="gpt-6-luna",
        max_tokens=256,
        messages=[{"role": "user", "content": "hi"}],
    )
    print(message.content[0].text)

The async clients work the same way inside async with:

async def main():
    async with rg.Gateway() as gw:
        client = gw.openai_async()
        reply = await client.chat.completions.create(
            model="claude-haiku-5-5",
            messages=[{"role": "user", "content": "hi"}],
        )

Streaming works as it does against the real API: pass stream=True and iterate.

Each method takes api_key (see Client keys) and passes any other keyword arguments to the SDK client. Leaving the with or async with block stops any servers the gateway started and closes the clients it handed out.

Other httpx clients

For any client built on httpx, use the gateway’s transports:

import httpx

http = httpx.Client(transport=gw.transport())
ahttp = httpx.AsyncClient(transport=gw.async_transport())

Raw requests

gw.request sends one request and returns an httpx.Response. Its keyword arguments are those of httpx.Client.request (json, content, files, headers, params, timeout).

response = gw.request("POST", "/v1/chat/completions", json={
    "model": "claude-haiku-5-5",
    "messages": [{"role": "user", "content": "hi"}],
})
print(response.status_code, response.json())

gw.stream yields the response body chunk by chunk as it arrives, which is what you want for server-sent events:

for chunk in gw.stream("POST", "/v1/chat/completions", json={
    "model": "claude-haiku-5-5",
    "stream": True,
    "messages": [{"role": "user", "content": "hi"}],
}):
    print(chunk.decode(), end="")

gw.arequest and gw.astream are the asyncio versions. There is no timeout unless you pass one.

Images, audio, embeddings and video

The OpenAI client also covers the other kinds of request, each served when a provider speaks its format (openai-images, openai-transcriptions, openai-speech, openai-embeddings, openai-videos). See Providers and models for which vendors have which.

client = gw.openai()

# Image
client.images.generate(model="gpt-image-1", prompt="a red fox").data[0].b64_json

# Transcription (a multipart upload)
with open("talk.wav", "rb") as f:
    client.audio.transcriptions.create(model="whisper-1", file=f).text

# Speech: audio bytes
client.audio.speech.create(model="tts-1", voice="alloy", input="hi").read()

# Embeddings
client.embeddings.create(model="text-embedding-3-small", input="hi").data[0].embedding

# Video (Veo): create, poll, download
job = client.videos.create(model="veo-3.0-generate-001", prompt="waves")
job = client.videos.retrieve(job.id)             # queued, then completed
client.videos.download_content(job.id).read()    # mp4 bytes

What the gateway knows

gw.models()                     # every provider's models, merged, as GET /v1/models lists them
gw.models(kind="image")         # only image models
gw.providers()                  # each provider with its formats and request counts
gw.requests(limit=20)           # the request log, newest first
gw.requests(client="ci")        # only requests made with the client key named "ci"
gw.export_requests("csv")       # the whole log as CSV (or "jsonl")

Each model is a dict with id, owned_by (the provider) and kind: chat, embedding, image, transcription, speech, video or realtime. gw.models() asks every provider, so it takes as long as the slowest one.

requests() and export_requests() also filter by kind, status (ok, error, pending, or a class such as 2xx), provider, model and since.

providers(), requests() and export_requests() read the web UI’s admin API in-process, so ui = false in the config turns them off. These helpers never need a client key. amodels(), aproviders(), arequests() and aexport_requests() are the asyncio versions.

The request log is kept in memory. A gateway in your process prints nothing per request, because log defaults to off here (the binary’s default is text). Set log = "text" or log = "json" in the config to get stderr lines.

Errors

• An HTTP error status is a response, not an exception of this package. The SDKs raise their own exceptions for it, as usual.
• A failure to carry out a request or start a server raises rg.GatewayError.
• A timeout raises httpx.ReadTimeout, which an SDK turns into its own timeout error.
• An invalid configuration raises rg.ConfigError, a subclass of ValueError.

try:
    gw = rg.Gateway({"surprise": 1})        # not a config setting
except rg.ConfigError as e:
    print(e)

Serve on a socket

Some clients need a real URL. serve() starts the gateway’s HTTP server in the background and returns a Server with .url and .stop():

with gw.serve() as server:          # 127.0.0.1, a free port
    print(server.url)               # http://127.0.0.1:54321
    print(server.url + "/ui/")      # the web UI

async with gw.aserve() as server:   # the same for asyncio
    ...

gw.serve_forever(port=8080)         # blocks; Ctrl-C stops it

serve(host="127.0.0.1", port=0) picks a free port by default; serve_forever defaults to port 8080 and prints the same banner as the binary. A server from serve() keeps running until you stop it, the gateway is closed, or the process exits, even if you drop the Server.

The web UI is included in wheels from PyPI. A build from the sdist or a plain checkout shows a placeholder page unless apps/ui was built first.

Realtime

WebSocket sessions (OpenAI Realtime and Gemini Live, translated between each other) need a socket. Use serve() and connect any WebSocket client, with http replaced by ws:

import websockets

with gw.serve() as server:
    url = server.url.replace("http", "ws") + "/v1/realtime?model=gpt-realtime"
    async with websockets.connect(url) as ws:
        ...

Configuration

Gateway(config=...) takes a dict or a TOML string, in the same format as gateway.toml. See Configuration for every setting.

gw = rg.Gateway({
    "providers": {
        "openai": {"vendor": "openai", "key": "env:OPENAI_API_KEY"},
        "anthropic": {"vendor": "anthropic", "key": "env:ANTHROPIC_API_KEY"},
    },
    "routes": {
        "smart": {"to": ["anthropic/claude-haiku-5-5", "openai/gpt-6-luna"], "tries": 2},
    },
})

gw = rg.Gateway("""
[providers.openai]
vendor = "openai"
key = "env:OPENAI_API_KEY"

[routes.smart]
to = ["openai/gpt-6-luna"]
""")

gw = rg.Gateway.from_file("gateway.toml")

With no config, rg.Gateway() starts the way the binary does with no arguments: from ./gateway.toml if that file exists, otherwise from the environment. Every vendor whose key is set (OPENAI_API_KEY, ANTHROPIC_API_KEY, GEMINI_API_KEY, GROQ_API_KEY, …) becomes a provider, and RESPAN_<VENDOR>_BASE_URL moves a vendor’s base URL. See Providers and models for how a model name picks a provider.

Client keys

If the config sets client_keys, every request needs one, in-process requests included. Pass it to the client:

import os

gw = rg.Gateway({
    "client_keys": [{"name": "app", "key": "env:APP_KEY"}],
    "providers": {"openai": {"vendor": "openai", "key": "env:OPENAI_API_KEY"}},
})
client = gw.openai(api_key=os.environ["APP_KEY"])

Without client keys, api_key is a placeholder. It reaches a provider only if the gateway holds no key of its own for it.

Command line

The package installs the same command line as the binary:

respan-gateway                    # also: init, check, models, mcp, proxy --anthropic|--openai
python -m respan_gateway proxy --openai

See CLI.