Wrapping a python requests llm rest api call around a hosted inference gateway is the fastest path from prototype to production for many backend services. The n4n.ai gateway exposes a single OpenAI-compatible REST endpoint that fronts 240+ models and handles provider fallback transparently, so your python requests llm rest api client can stay simple. This guide walks through the exact steps to authenticate, send chat completions, stream tokens, and handle errors with the standard requests library.
Step 1: Export your API key and configure the environment
Never hardcode credentials. Pull them from the environment so the same code runs locally and in CI.
import os
API_KEY = os.environ.get("N4N_API_KEY")
if not API_KEY:
raise RuntimeError("Set N4N_API_KEY in your environment")
BASE_URL = "https://api.n4n.ai/v1/chat/completions"
Set the variable in your shell before running:
export N4N_API_KEY="sk-..."
If you run in containers, inject the secret via your orchestrator’s secret store, not via a committed .env file.
Step 2: Install requests and define a minimal client
The requests library is still the most debuggable HTTP client for server-side Python. Install it explicitly to avoid relying on transitive deps.
pip install requests==2.31.0
Create a helper that builds the auth header. The gateway uses bearer tokens exactly like the OpenAI API.
import requests
def headers() -> dict:
return {
"Authorization": f"Bearer {API_KEY}",
"Content-Type": "application/json",
}
For production, wrap the client in a requests.Session to reuse connections:
session = requests.Session()
session.headers.update(headers())
Step 3: Discover available models
Before hardcoding a model string, query the catalog. The gateway is OpenAI-compatible, so a GET /v1/models works.
models_url = BASE_URL.replace("/chat/completions", "/models")
resp = session.get(models_url, timeout=10)
resp.raise_for_status()
models = resp.json()["data"]
print([m["id"] for m in models[:5]])
This confirms your key is valid and shows the exact model IDs. The python requests llm rest api client now has a dynamic menu instead of a guess.
Step 4: Send a non-streaming chat completion
A basic call mirrors the OpenAI schema. Pick a model from the 240+ available by name; the gateway routes it to the correct provider.
payload = {
"model": "anthropic/claude-3-haiku",
"messages": [
{"role": "system", "content": "You are a terse code reviewer."},
{"role": "user", "content": "Review: def foo(): return 1"}
],
"temperature": 0.2,
"max_tokens": 256,
}
resp = session.post(BASE_URL, json=payload, timeout=30)
resp.raise_for_status()
data = resp.json()
print(data["choices"][0]["message"]["content"])
The call is synchronous here; it blocks until the full completion returns. Use this mode for batch jobs or tests.
Step 5: Parse usage and confirm per-token metering
The response includes a usage object. The gateway meters per token, so log it for cost attribution.
usage = data.get("usage", {})
print(f"prompt_tokens={usage.get('prompt_tokens')} "
f"completion_tokens={usage.get('completion_tokens')} "
f"total_tokens={usage.get('total_tokens')}")
If you batch requests, aggregate these fields in your own metrics pipeline. The gateway does not hide provider token counts.
Step 6: Stream tokens with stream=True
For chat UIs, stream to reduce time-to-first-token. requests supports SSE via iter_lines.
import json
stream_payload = {**payload, "stream": True}
with session.post(BASE_URL, json=stream_payload, stream=True, timeout=30) as r:
r.raise_for_status()
for line in r.iter_lines():
if not line:
continue
line = line.decode("utf-8")
if line.startswith("data:"):
event = line[len("data:"):].strip()
if event == "[DONE]":
break
chunk = json.loads(event)
delta = chunk["choices"][0]["delta"].get("content", "")
if delta:
print(delta, end="", flush=True)
You must consume the stream fully or the connection leaks. Use a with block as shown.
Step 7: Forward cache-control and routing hints
The gateway honors client routing directives and forwards provider cache-control hints. To leverage provider-side prompt caching, send a standard HTTP Cache-Control header.
cached_headers = headers()
cached_headers["Cache-Control"] = "max-age=300"
resp = session.post(BASE_URL, headers=cached_headers, json=payload, timeout=30)
Routing preferences (e.g., avoid a degraded provider) are passed via gateway-specific headers or body fields; consult the docs for the exact key name. The python requests llm rest api call itself does not change—only the headers do.
Step 8: Handle errors, timeouts, and retries
Network failures are inevitable. Wrap calls in a retry loop with exponential backoff. The gateway already performs automatic fallback when a provider is rate-limited, but a 429 from the gateway means you are hitting global limits.
from time import sleep
def post_with_retry(payload, attempts=3):
for i in range(attempts):
try:
r = session.post(BASE_URL, json=payload, timeout=30)
if r.status_code == 429:
sleep(2 ** i)
continue
r.raise_for_status()
return r.json()
except requests.RequestException as e:
if i == attempts - 1:
raise
sleep(2 ** i)
Set timeout always. Without it, a stalled connection can hang a worker indefinitely. Distinguish 400 (bad request) from 401 (auth) from 5xx (gateway/provider) in logs.
Step 9: Verify success end to end
A robust verification is to assert the response shape and nonzero token usage.
def verify_completion(data):
assert "choices" in data and data["choices"]
assert data["choices"][0]["finish_reason"] in ("stop", "length")
assert data["usage"]["total_tokens"] > 0
return True
# After a call:
verify_completion(data)
print("OK")
Run the script. If you see streamed tokens or a printed review with total_tokens > 0, the integration works. For streaming, check that [DONE] arrives and finish_reason is present in the final chunk.
Production hardening notes
- Use a
requests.Sessionto reuse TCP connections across calls. - Disable
urllib3retry at the adapter level if you implement your own backoff to avoid double retries. - Log the
X-Request-Idresponse header (if provided) for support tickets. - Never log raw completions in shared environments; redact as needed.
- For async services, switch to
httpxwith the same JSON contract—the request shape does not change.
The python requests llm rest api pattern above is deliberately low-level. It avoids SDK magic so you can debug headers, timeouts, and payloads directly. When you outgrow raw calls, layer a thin wrapper that enforces your retry and logging policies, but keep the wire format identical to what we used here.