---
title: "Python, in your code"
subtitle: "Run the gateway inside your Python process and call it with the OpenAI or Anthropic SDK. No server, no socket."
---

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

```python
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

```bash
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.

| Method | Returns |
| --- | --- |
| `gw.openai()` | `openai.OpenAI` |
| `gw.openai_async()` | `openai.AsyncOpenAI` |
| `gw.anthropic()` | `anthropic.Anthropic` |
| `gw.anthropic_async()` | `anthropic.AsyncAnthropic` |

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

```python
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`:

```python
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](#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:

```python
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`).

```python
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:

```python
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](/docs/oss-gateway/providers) for which vendors have which.

```python
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

```python
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`.

```python
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()`:

```python
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`.

<Warning>
Without `client_keys` in the config, anyone who can reach the server spends your provider keys. Keep it on `127.0.0.1`, or set [client keys](/docs/oss-gateway/client-keys) before you serve on another address.
</Warning>

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`:

```python
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](/docs/oss-gateway/config) for every setting.

<Tabs>
<Tab title="Dict">
```python
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},
    },
})
```

`rg.GatewayConfig` types the dict, for editor completion and type checking.
</Tab>
<Tab title="TOML string">
```python
gw = rg.Gateway("""
[providers.openai]
vendor = "openai"
key = "env:OPENAI_API_KEY"

[routes.smart]
to = ["openai/gpt-6-luna"]
""")
```
</Tab>
<Tab title="File">
```python
gw = rg.Gateway.from_file("gateway.toml")
```
</Tab>
</Tabs>

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](/docs/oss-gateway/providers) 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:

```python
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:

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

See [CLI](/docs/oss-gateway/cli).