Install langchain-openai and route requests through n4n.ai

The langchain-openai package is the standard way to call OpenAI-compatible APIs from LangChain. Since n4n.ai exposes an OpenAI-compatible endpoint, you can point the client at the gateway and immediately get access to 240+ models, automatic provider fallback, and per-token usage metering without changing your application logic. This guide walks through installation, configuration, and a minimal verification script you can run in under five minutes.

Step 1: Install the package

Use your preferred package manager. The examples below assume Python 3.10+ and a virtual environment.

pip install langchain-openai

If you use Poetry:

poetry add langchain-openai

For uv:

uv add langchain-openai

Verify the import works:

# verify_import.py
from langchain_openai import ChatOpenAI
print("Import successful")

Run it:

python verify_import.py

You should see Import successful with no errors.

Step 2: Create an n4n.ai API key

You need an API key from the n4n.ai dashboard. Log in, navigate to API Keys, and create a new key. Copy it — you won’t see it again.

Store the key in your environment rather than hardcoding it:

export N4N_API_KEY="sk-n4n-..."

On Windows PowerShell:

$env:N4N_API_KEY = "sk-n4n-..."

Step 3: Configure the ChatOpenAI client

The ChatOpenAI class accepts base_url and api_key parameters. Point base_url to the n4n.ai gateway endpoint and pass your key.

# basic_chat.py
import os
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(
    model="gpt-4o-mini",  # any model name n4n.ai supports
    api_key=os.getenv("N4N_API_KEY"),
    base_url="https://api.n4n.ai/v1",
    temperature=0.2,
)

response = llm.invoke([HumanMessage(content="Say 'routed through gateway' in exactly three words.")])
print(response.content)

Run it:

python basic_chat.py

Expected output (or similar):

Routed through gateway

The request traveled: your code → langchain-openai → n4n.ai gateway → upstream provider → back through the gateway → your code.

Step 4: Verify routing with response headers

n4n.ai returns headers that confirm which upstream provider handled the request and whether fallback occurred. The ChatOpenAI client doesn’t expose raw headers by default, so use the lower-level openai client for inspection, or enable LangChain’s callback handler.

Option A: Raw OpenAI client (simplest verification)

# verify_routing.py
import os
from openai import OpenAI

client = OpenAI(
    api_key=os.getenv("N4N_API_KEY"),
    base_url="https://api.n4n.ai/v1",
)

completion = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "ping"}],
    max_tokens=5,
)

print("Response:", completion.choices[0].message.content)
print("Headers:")
for k, v in completion._response.headers.items():
    if k.lower().startswith(("x-", "cf-", "via")):
        print(f"  {k}: {v}")

Run it and look for headers like x-n4n-provider, x-n4n-model, or x-n4n-fallback. Their presence confirms the request passed through the gateway.

Option B: LangChain callback for structured metadata

If you prefer staying in LangChain, use a custom callback handler to capture the response metadata that n4n.ai injects.

# callback_verify.py
import os
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain_core.callbacks import BaseCallbackHandler
from typing import Any, Dict

class GatewayMetadataHandler(BaseCallbackHandler):
    def on_llm_end(self, response: Any, **kwargs: Any) -> None:
        # n4n.ai attaches metadata to the generation's response_metadata
        for gen in response.generations[0]:
            meta = gen.message.response_metadata
            if meta:
                print("Gateway metadata:")
                for k, v in meta.items():
                    if k.startswith("n4n_"):
                        print(f"  {k}: {v}")

llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("N4N_API_KEY"),
    base_url="https://api.n4n.ai/v1",
    callbacks=[GatewayMetadataHandler()],
    temperature=0,
)

llm.invoke([HumanMessage(content="pong")])

Run it. You should see metadata fields like n4n_provider, n4n_model, and optionally n4n_fallback_reason if the primary provider was unhealthy.

Step 5: Use routing directives for model selection

n4n.ai honors client-side routing directives via the model parameter and optional headers. You can request a specific provider or capability without changing code paths.

Specify a provider family

# provider_routing.py
import os
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

# Request an Anthropic model via the gateway
llm = ChatOpenAI(
    model="claude-3-5-sonnet-20241022",  # n4n.ai maps this to Anthropic
    api_key=os.getenv("N4N_API_KEY"),
    base_url="https://api.n4n.ai/v1",
    temperature=0.3,
)

response = llm.invoke([HumanMessage(content="What provider am I using?")])
print(response.content)

Request capabilities instead of a specific model

Use n4n.ai’s capability-based routing by passing a model alias like best-for-coding or cheapest-available. The gateway selects the optimal provider at request time.

# capability_routing.py
import os
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(
    model="best-for-coding",  # capability alias
    api_key=os.getenv("N4N_API_KEY"),
    base_url="https://api.n4n.ai/v1",
    temperature=0.1,
)

response = llm.invoke([HumanMessage(content="Write a Python decorator that retries a function 3 times with exponential backoff.")])
print(response.content)

Check the n4n_provider metadata to see which model actually served the request.

Step 6: Handle streaming responses

Streaming works identically to the standard OpenAI client. Use stream() for token-by-token output.

# streaming.py
import os
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("N4N_API_KEY"),
    base_url="https://api.n4n.ai/v1",
    temperature=0.5,
    streaming=True,
)

for chunk in llm.stream([HumanMessage(content="Count from 1 to 10, one number per line.")]):
    print(chunk.content, end="", flush=True)
print()

The gateway streams tokens from the upstream provider without buffering. If the primary provider degrades mid-stream, n4n.ai attempts fallback — though note that mid-stream fallback may produce a visible seam in the output.

Step 7: Configure timeouts and retries

Production code needs explicit timeout and retry settings. The ChatOpenAI constructor accepts timeout, max_retries, and request_timeout (alias for timeout).

# production_config.py
import os
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("N4N_API_KEY"),
    base_url="https://api.n4n.ai/v1",
    temperature=0.2,
    timeout=30.0,           # total request timeout in seconds
    max_retries=2,          # client-side retries for transient errors
)

response = llm.invoke([HumanMessage(content="Health check")])
print("OK:", response.content[:50])

The gateway itself enforces its own upstream timeouts and retry logic. Setting client-side values prevents your application from hanging if the gateway becomes unreachable.

Step 8: Meter usage in your application

n4n.ai returns usage metadata in the standard OpenAI format (prompt_tokens, completion_tokens, total_tokens). Capture it for cost tracking or rate limiting.

# usage_metering.py
import os
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("N4N_API_KEY"),
    base_url="https://api.n4n.ai/v1",
    temperature=0,
)

response = llm.invoke([HumanMessage(content="Summarize the plot of Hamlet in one sentence.")])

usage = response.response_metadata.get("token_usage", {})
print(f"Prompt tokens: {usage.get('prompt_tokens')}")
print(f"Completion tokens: {usage.get('completion_tokens')}")
print(f"Total tokens: {usage.get('total_tokens')}")

For streaming, aggregate usage from the final chunk:

# streaming_usage.py
import os
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

llm = ChatOpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("N4N_API_KEY"),
    base_url="https://api.n4n.ai/v1",
    temperature=0.3,
    streaming=True,
)

final_chunk = None
for chunk in llm.stream([HumanMessage(content="Write a haiku about distributed systems.")]):
    print(chunk.content, end="", flush=True)
    final_chunk = chunk
print()

if final_chunk and final_chunk.response_metadata:
    usage = final_chunk.response_metadata.get("token_usage", {})
    print(f"\nTotal tokens: {usage.get('total_tokens')}")

Step 9: Run a quick integration test

Combine the pieces into a single script that validates the full path: authentication, routing, streaming, and usage capture.

# integration_test.py
"""Run: python integration_test.py"""
import os
import sys
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

API_KEY = os.getenv("N4N_API_KEY")
if not API_KEY:
    print("ERROR: Set N4N_API_KEY environment variable", file=sys.stderr)
    sys.exit(1)

def test_non_streaming():
    llm = ChatOpenAI(
        model="gpt-4o-mini",
        api_key=API_KEY,
        base_url="https://api.n4n.ai/v1",
        temperature=0,
        max_tokens=20,
    )
    resp = llm.invoke([HumanMessage(content="Reply with exactly: non-streaming ok")])
    assert "non-streaming ok" in resp.content.lower()
    usage = resp.response_metadata.get("token_usage", {})
    assert usage.get("total_tokens", 0) > 0
    print("✓ Non-streaming works")
    return True

def test_streaming():
    llm = ChatOpenAI(
        model="gpt-4o-mini",
        api_key=API_KEY,
        base_url="https://api.n4n.ai/v1",
        temperature=0,
        max_tokens=20,
        streaming=True,
    )
    collected = []
    for chunk in llm.stream([HumanMessage(content="Reply with exactly: streaming ok")]):
        collected.append(chunk.content)
    full = "".join(collected).lower()
    assert "streaming ok" in full
    print("✓ Streaming works")
    return True

def test_capability_routing():
    llm = ChatOpenAI(
        model="cheapest-available",
        api_key=API_KEY,
        base_url="https://api.n4n.ai/v1",
        temperature=0,
        max_tokens=10,
    )
    resp = llm.invoke([HumanMessage(content="ping")])
    meta = resp.response_metadata
    provider = meta.get("n4n_provider") or meta.get("model_name")
    print(f"✓ Capability routing works (provider: {provider})")
    return True

if __name__ == "__main__":
    test_non_streaming()
    test_streaming()
    test_capability_routing()
    print("\nAll integration checks passed.")

Run it:

python integration_test.py

Expected output:

✓ Non-streaming works
✓ Streaming works
✓ Capability routing works (provider: openai/gpt-4o-mini)

All integration checks passed.

Step 10: Common pitfalls and fixes

Next steps

You now have a working LangChain → n4n.ai integration. From here:

• Add observability: Wrap calls with LangSmith or your preferred tracer; the gateway’s n4n_* metadata fields correlate traces with provider selection.
• Implement fallback policies: Configure provider priority lists in the n4n.ai dashboard rather than hardcoding model names.
• Set budget alerts: Use the per-token usage data to enforce cost ceilings per workflow or tenant.
• Explore async: Switch to ChatOpenAI(...).ainvoke() and astream() for high-concurrency workloads.

The gateway handles provider complexity so your application code stays clean. Change models, add providers, or adjust routing rules in the dashboard — no redeploy required.