Python SDK
The official nRouter Python SDK with automatic metadata extraction, real-time cost tracking, Anthropic messages, and context management.
Last updated
The official nRouter Python SDK (nrouter-sdk) extends the standard OpenAI interface with automatic x-nr-* header parsing, real-time USD cost calculation, native Anthropic Messages API support, server-side prompt templating, and connection pool lifecycle management.
Installation
Install the official package from PyPI:
pip install nrouter-sdkAuthentication & Setup
Export your virtual API key in your environment:
export NROUTER_API_KEY="sk-nrouter-your-key-here"Initialize the client with automatic environment resolution and context management:
from nroutersdk import nRouter
# Initialize client — reads NROUTER_API_KEY and NROUTER_BASE_URL from env
with nRouter() as client:
response = client.chat.completions.create(
model="gpt-5.4-mini",
messages=[{"role": "user", "content": "Hello, nRouter!"}],
)
print(response.choices[0].message.content)
# Automatic response metadata extraction
if client.last_response:
print(f"Request ID: {client.last_response.request_id}")
print(f"Cost (USD): ${client.last_response.cost}")
print(f"Cache Hit: {client.last_response.is_cache_hit}")Core Capabilities & Examples
| Topic | Example File | Description |
|---|---|---|
| Quickstart | 01_quickstart.py | Basic chat completion & automatic metadata extraction |
| Async / Concurrency | 02_async_concurrency.py | Concurrent requests with asyncio.gather |
| Streaming | 03_streaming.py | Server-Sent Events (SSE) token streaming |
| Anthropic Messages | 04_anthropic_messages.py | Native Anthropic /v1/messages format & token counting |
| Metadata & Cost | 05_metadata_cost_tracking.py | Inspect x-nr-* headers and cost attribution |
| Prompt Templates | 06_prompt_templates.py | Server-side prompt templates & variable injection |
| Tool Calling | 07_tool_calling.py | Function calling with JSON schema tools |
| Structured Outputs | 08_structured_outputs.py | Strict JSON object output formatting |
| Error Handling | 09_error_handling.py | Typed error handling & guardrail recovery |
| Memory | 10_conversation_memory.py | Multi-turn conversation memory |
| Embeddings | 11_embeddings.py | Vector embeddings generation |
| Multimodal Vision | 12_multimodal_vision.py | Image inputs via URL or base64 |
Interactive notebook available at notebooks/quickstart.ipynb.
1. Streaming Responses
Stream tokens in real-time using Server-Sent Events (SSE):
from nroutersdk import nRouter
client = nRouter()
stream = client.chat.completions.create(
model="gpt-5.4-mini",
messages=[{"role": "user", "content": "Write a 3-paragraph summary of quantum computing."}],
stream=True,
)
for chunk in stream:
delta = chunk.choices[0].delta.content
if delta:
print(delta, end="", flush=True)
print()2. Async & High Concurrency
Scale IO-bound workloads with AsyncnRouter and asyncio.gather:
import asyncio
from nroutersdk import AsyncnRouter
async def fetch_summary(topic: str, client: AsyncnRouter):
response = await client.chat.completions.create(
model="gpt-5.4-mini",
messages=[{"role": "user", "content": f"Summarize {topic} in one sentence."}],
)
return topic, response.choices[0].message.content
async def main():
topics = ["Kubernetes", "PostgreSQL", "Redis", "Kafka", "Rust"]
async with AsyncnRouter() as client:
tasks = [fetch_summary(t, client) for t in topics]
results = await asyncio.gather(*tasks)
for topic, summary in results:
print(f"[{topic}]: {summary}")
asyncio.run(main())3. Native Anthropic Messages API
Use the native Anthropic Messages API format across any provider without changing SDKs:
from nroutersdk import nRouter
with nRouter() as client:
# 1. Count tokens before execution
token_count = client.messages.count_tokens(
model="claude-sonnet-4-5-20250929",
messages=[{"role": "user", "content": "Analyze this 500-line log file..."}],
)
print(f"Estimated input tokens: {token_count['input_tokens']}")
# 2. Execute Messages request
response = client.messages.create(
model="claude-sonnet-4-5-20250929",
max_tokens=1024,
messages=[{"role": "user", "content": "Explain vector indexing simply."}],
)
print(response['content'][0]['text'])4. Server-Side Prompt Templates
Inject version-controlled templates and variables server-side without prompt leaks:
from nroutersdk import nRouter
with nRouter() as client:
response = client.nrouter.chat(
model="gpt-5.4-mini",
prompt_template_id="tmpl_support_ticket_v2",
prompt_variables={
"customer_tier": "Enterprise",
"issue_category": "Billing Discrepancy",
},
messages=[{"role": "user", "content": "User reports double charge on invoice #9021."}],
)
print(response.choices[0].message.content)5. Tool & Function Calling
Provide typed tools using JSON Schema:
import json
from nroutersdk import nRouter
tools = [
{
"type": "function",
"function": {
"name": "get_stock_price",
"description": "Fetch real-time stock ticker price",
"parameters": {
"type": "object",
"properties": {
"ticker": {"type": "string", "description": "Stock symbol (e.g. AAPL, GOOG)"}
},
"required": ["ticker"],
},
},
}
]
with nRouter() as client:
response = client.chat.completions.create(
model="gpt-5.4-mini",
messages=[{"role": "user", "content": "What is the current stock price of Apple?"}],
tools=tools,
tool_choice="auto",
)
message = response.choices[0].message
if message.tool_calls:
for tool_call in message.tool_calls:
args = json.loads(tool_call.function.arguments)
print(f"Execute function {tool_call.function.name} with args: {args}")6. Typed Error Handling & Guardrails
Recover cleanly from network timeouts, quota limits, and guardrail redactions:
from nroutersdk import (
nRouter,
nRouterError,
nRouterAuthenticationError,
nRouterRateLimitError,
nRouterGuardrailBlockedError,
)
client = nRouter()
try:
response = client.chat.completions.create(
model="claude-sonnet-4-5-20250929",
messages=[{"role": "user", "content": "Process sensitive transaction..."}],
)
except nRouterGuardrailBlockedError as e:
print(f"Request blocked by safety policy: {e}")
except nRouterRateLimitError as e:
print(f"Rate limited: retry after {e.retry_after}s")
except nRouterAuthenticationError:
print("Invalid API key — verify NROUTER_API_KEY environment variable.")
except nRouterError as e:
print(f"Gateway error (HTTP {e.status_code}): {e.message}")
finally:
client.close()