Python

Two patterns are common:

1. Use the official OpenAI SDK and point it at the gateway.
2. Use httpx (or requests) directly when you need full control over headers, streaming, or specialized metadata.

Option 1 — OpenAI SDK

pip install openai

import os
from openai import OpenAI

client = OpenAI(
    base_url=os.environ["SMART_GATEWAY_BASE_URL"] + "/v1",
    api_key=os.environ["SMART_GATEWAY_API_KEY"],
)

resp = client.chat.completions.create(
    model="smart-router",
    messages=[{"role": "user", "content": "Reply with PONG."}],
    max_tokens=8,
)

print(resp.choices[0].message.content)

Streaming

stream = client.chat.completions.create(
    model="smart-router-flash",
    stream=True,
    messages=[{"role": "user", "content": "Count to 5."}],
    max_tokens=64,
)

for chunk in stream:
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)
print()

Sending custom metadata

resp = client.chat.completions.create(
    model="smart-router",
    messages=[{"role": "user", "content": "..."}],
    extra_body={
        "metadata": {
            "sensitive": True,
            "reasoning": "high",
            "session_id": "proj-A-2026-08-23",
            "goal_id": "fix-flaky",
            "task_id": "task-17",
        }
    },
)

Sending custom headers

The OpenAI SDK passes custom headers via default_headers:

client = OpenAI(
    base_url=os.environ["SMART_GATEWAY_BASE_URL"] + "/v1",
    api_key=os.environ["SMART_GATEWAY_API_KEY"],
    default_headers={
        "X-Smart-Gateway-Session": "proj-A-2026-08-23",
        "X-Smart-Gateway-Goal": "fix-flaky",
        "X-Task-Id": "task-17",
        "X-Pro": "true",
    },
)

Option 2 — httpx

Useful when you want byte-level control, e.g. for SSE with custom parsing.

import os, json, httpx

BASE = os.environ["SMART_GATEWAY_BASE_URL"]
KEY = os.environ["SMART_GATEWAY_API_KEY"]

def chat(messages, model="smart-router", **kw):
    payload = {"model": model, "messages": messages, **kw}
    headers = {
        "Authorization": f"Bearer {KEY}",
        "Content-Type": "application/json",
    }
    r = httpx.post(f"{BASE}/v1/chat/completions", headers=headers, json=payload, timeout=60.0)
    r.raise_for_status()
    return r.json()

resp = chat([{"role": "user", "content": "PONG"}], model="smart-router-flash", max_tokens=4)
print(resp["choices"][0]["message"]["content"])

Streaming with httpx

import httpx

payload = {
    "model": "smart-router-flash",
    "stream": True,
    "messages": [{"role": "user", "content": "Count to 5."}],
    "max_tokens": 64,
}
headers = {
    "Authorization": f"Bearer {KEY}",
    "Content-Type": "application/json",
}

with httpx.stream("POST", f"{BASE}/v1/chat/completions", headers=headers, json=payload, timeout=60.0) as r:
    r.raise_for_status()
    for line in r.iter_lines():
        if not line or not line.startswith("data: "):
            continue
        data = line[6:]
        if data == "[DONE]":
            break
        chunk = json.loads(data)
        delta = chunk["choices"][0]["delta"].get("content") or ""
        if delta:
            print(delta, end="", flush=True)
print()

Async (httpx + asyncio)

import asyncio, httpx, os

BASE = os.environ["SMART_GATEWAY_BASE_URL"]
KEY = os.environ["SMART_GATEWAY_API_KEY"]

async def main():
    async with httpx.AsyncClient(timeout=60.0) as client:
        r = await client.post(
            f"{BASE}/v1/chat/completions",
            headers={"Authorization": f"Bearer {KEY}", "Content-Type": "application/json"},
            json={"model": "smart-router-flash",
                  "messages": [{"role": "user", "content": "PONG"}],
                  "max_tokens": 4},
        )
        r.raise_for_status()
        print(r.json()["choices"][0]["message"]["content"])

asyncio.run(main())

Robust retry wrapper

import time, httpx

def chat_with_retry(payload, *, max_attempts=3, base_delay=1.0):
    headers = {"Authorization": f"Bearer {KEY}", "Content-Type": "application/json"}
    for attempt in range(max_attempts):
        try:
            r = httpx.post(f"{BASE}/v1/chat/completions", headers=headers, json=payload, timeout=60.0)
            if r.status_code == 429 or r.status_code >= 500:
                raise httpx.HTTPStatusError("retryable", request=r.request, response=r)
            r.raise_for_status()
            return r.json()
        except (httpx.HTTPError, httpx.HTTPStatusError) as e:
            if attempt == max_attempts - 1:
                raise
            time.sleep(base_delay * (2 ** attempt))

This mirrors the gateway's own retry policy against OpenRouter (see src/smart_gateway/clients/openrouter.py), so the two layers cooperate instead of stacking up.