Running open-source agent models n4n.ai turns a fragmented multi-vendor setup into a single OpenAI-compatible call surface. You get Llama 4, Mistral, Qwen, DeepSeek, and Grok behind one base URL, with the gateway handling provider failover and per-token accounting so your agent loop doesn’t care which weight file actually executed the step.
Step 1: Point the OpenAI client at the gateway
Install the SDK and configure credentials. The n4n.ai endpoint speaks the OpenAI Chat Completions contract verbatim, so existing code only needs a base_url change.
export N4N_API_KEY="sk-..."
pip install openaiimport os
from openai import OpenAI
client = OpenAI(
base_url="https://api.n4n.ai/v1", # n4n.ai OpenAI-compatible endpoint
api_key=os.environ["N4N_API_KEY"],
)Verify connectivity with a raw curl before writing agent logic:
curl https://api.n4n.ai/v1/models \
-H "Authorization: Bearer $N4N_API_KEY" | head -c 200You should see a JSON list containing entries like meta-llama/llama-4-scout. If you get 401, the key is wrong; 404 means the path is off.
Step 2: Select a model and send a minimal prompt
Model identifiers follow org/name. The open-source agent models we care about:
meta-llama/llama-4-scout(long context)mistralai/mistral-large-2407(fast, stable tool use)qwen/qwen2.5-72b-instruct(multilingual)deepseek/deepseek-r1-distill(reasoning)x-ai/grok-2-mini(recent-data access)
resp = client.chat.completions.create(
model="meta-llama/llama-4-scout",
messages=[{"role": "user", "content": "List the steps to rotate an API key safely."}],
temperature=0.2,
)
print(resp.choices[0].message.content)
print(resp.usage.total_tokens)Success means a non-empty content and a usage object with prompt_tokens and completion_tokens populated. Do not skip the usage check; it confirms metering is live.
Step 3: Build a tool-calling agent loop
Open-source models differ in function-calling rigor. Llama 4 and Mistral emit clean parallel calls. Qwen and DeepSeek need strict schemas and sometimes wrap arguments in extra quotes. Grok expects JSON mode for reliable parses. Define one tool and run a loop that executes it.
tools = [{
"type": "function",
"function": {
"name": "get_service_status",
"description": "Return health of a service",
"parameters": {
"type": "object",
"properties": {"service": {"type": "string"}},
"required": ["service"],
},
},
}]
def run_agent(model, user_msg):
messages = [{"role": "user", "content": user_msg}]
for _ in range(5):
resp = client.chat.completions.create(
model=model, messages=messages, tools=tools, tool_choice="auto")
msg = resp.choices[0].message
messages.append(msg)
if not msg.tool_calls:
return msg.content
for call in msg.tool_calls:
args = call.function.arguments
# naive parse; production should validate with pydantic
result = {"status": "green"}
messages.append({"role": "tool",
"tool_call_id": call.id,
"content": str(result)})
return "loop exhausted"
print(run_agent("mistralai/mistral-large-2407", "Is billing up?"))If the model returns tool_calls, your parser must handle missing fields. Log the raw arguments string when it fails.
Model-specific caveats
- Llama 4: Supports 128k+ context. Feed entire repo diffs.
- Mistral: Lowest p50 latency in my tests; use for high-QPS guardrails.
- Qwen: Best non-English tool use; set
temperature=0for JSON. - DeepSeek: Emits chain-of-thought tokens; strip them before final answer.
- Grok: Needs
response_format={"type": "json_object"}for stable calls.
Step 4: Pass routing directives and cache hints
The gateway honors client routing directives and forwards provider cache-control hints. When you have a preferred provider or want to cache a long system prompt, use headers:
resp = client.chat.completions.create(
model="deepseek/deepseek-r1-distill",
messages=[
{"role": "system", "content": "You are a SRE agent." * 50},
{"role": "user", "content": "Explain rollback for v2."},
],
extra_headers={
"x-routing-preference": "relative-quality",
"x-cache-ttl": "300",
},
)x-routing-preference accepts cheapest, fastest, or relative-quality. x-cache-ttl tells the upstream to reuse the prefix for 300 seconds. For agents that replay the same system prompt across thousands of tickets, this cuts token cost sharply.
Step 5: Add client-side retry as defense in depth
Automatic fallback at the gateway covers provider degradation, but your process should still handle 429/5xx. Wrap the call:
import time
def complete_with_retry(model, messages, tries=3):
for i in range(tries):
try:
return client.chat.completions.create(
model=model, messages=messages,
extra_headers={"x-routing-preference": "fastest"})
except Exception as e:
if "429" in str(e) or "rate" in str(e).lower():
time.sleep(2 ** i)
continue
raise
raise RuntimeError("exhausted retries")Because the gateway already shifts traffic to a healthy provider, your retry often succeeds on the first downstream attempt without exponential backoff firing.
Step 6: Meter per-token usage and aggregate
Every response includes usage. Log it by model to reconcile later:
from collections import defaultdict
import json
ledger = defaultdict(int)
def log_usage(resp):
ledger[resp.model] += resp.usage.total_tokens
# optionally write to stdout for shipping to metrics
print(json.dumps({"model": resp.model, "tokens": resp.usage.total_tokens}))
resp = complete_with_retry("qwen/qwen2.5-72b-instruct",
[{"role": "user", "content": "ping"}])
log_usage(resp)The gateway meters per token; if you run a fleet, scrape these logs and group by resp.model to see which open-source agent model is actually doing the work.
Step 7: End-to-end verification with a forced tool path
Write a test that forces exactly one tool call and a final answer:
def test_loop():
messages = [{"role": "user", "content": "Call test tool then say done."}]
resp = complete_with_retry("meta-llama/llama-4-scout", messages, tools=tools)
msg = resp.choices[0].message
assert msg.tool_calls, "expected tool call"
for c in msg.tool_calls:
messages.append({"role": "tool", "tool_call_id": c.id, "content": "ok"})
final = complete_with_retry("meta-llama/llama-4-scout", messages)
assert "done" in final.choices[0].message.content.lower()
print("PASS")
test_loop()Run it. A pass means the model invoked the tool, consumed the result, and produced a termination string.
Step 8: Stream tokens for interactive agents
For CLI or chat UIs, stream to avoid dead air:
stream = client.chat.completions.create(
model="mistralai/mistral-large-2407",
messages=[{"role": "user", "content": "Summarize this ticket"}],
stream=True,
)
for chunk in stream:
if chunk.choices[0].delta.content:
print(chunk.choices[0].delta.content, end="")Streaming works identically across all open-source agent models on the endpoint; no model-specific branching required.
Verify success
Your integration is complete when all of the following hold:
curl /v1/modelsreturns the open model IDs.- A non-streaming call returns
usage.total_tokens > 0. - Tool calls parse to valid arguments on at least Mistral and Llama 4.
- Forcing a 429 (e.g., low quota) triggers retry and recovers.
- The end-to-end
test_loop()prints PASS.
At that point you can swap any model= string and the agent keeps running. That portability is the whole point of routing open-source agent models through a single gateway instead of hard-coding provider URLs.
Practical model selection
Pick by failure mode, not leaderboard scores. If your agent breaks on long transcripts, Llama 4’s context wins. If latency budgets are tight, Mistral. If you serve EU users with non-English queries, Qwen. If the task needs multi-step planning with self-correction, DeepSeek’s reasoning trace helps despite verbosity. Grok fills the niche of fresh data without leaving the same API shape.
Because the request and response schemas are identical, A/B testing is a config change. Run both Mistral and Qwen on the same traffic for a day, compare ledger totals and tool-call success rates, then pin the winner.
That’s the setup. No vendor SDKs, no custom retry meshes, just one client and a model string.