LlamaIndex OpenAIAgent with n4n.ai function calling

The llamaindex openaiagent n4n.ai tutorial you are about to follow shows how to wire a LlamaIndex OpenAIAgent to an OpenAI-compatible gateway that speaks 240-plus models, surfaces provider cache-control hints, and falls back automatically when a provider is rate-limited or degraded. You will end up with a runnable agent that calls tools, streams responses, and reports per-token usage without any vendor-specific code in your application logic.

Prerequisites

Python 3.10 or newer
An n4n.ai API key (get one at https://n4n.ai)
Familiarity with LlamaIndex core concepts: Tool, AgentRunner, and the FunctionCallingAgent abstraction

Install the minimal dependency set:

pip install "llama-index-agent-openai>=0.2.0" "llama-index-core>=0.10.0" python-dotenv

Create a .env file in your project root:

N4N_API_KEY=sk-...
N4N_BASE_URL=https://api.n4n.ai/v1

The base URL is the only n4n.ai-specific configuration your application needs. Everything else is standard OpenAI-compatible surface area.

Define the tools

LlamaIndex agents expect tools that implement BaseTool or the simpler FunctionTool wrapper. We will build two tools: a deterministic calculator and a weather lookup that hits a public API. Both return structured data so the agent can reason over typed outputs.

# tools.py
import json
import os
import requests
from typing import Any
from llama_index.core.tools import FunctionTool

def calculate(expression: str) -> dict[str, Any]:
    """Evaluate a basic arithmetic expression safely."""
    allowed = set("0123456789+-*/(). ")
    if not set(expression).issubset(allowed):
        return {"error": "Expression contains disallowed characters"}
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return {"result": result}
    except Exception as e:
        return {"error": str(e)}

def get_weather(latitude: float, longitude: float) -> dict[str, Any]:
    """Fetch current weather from Open-Meteo (no key required)."""
    url = "https://api.open-meteo.com/v1/forecast"
    params = {
        "latitude": latitude,
        "longitude": longitude,
        "current_weather": "true",
        "timezone": "auto",
    }
    resp = requests.get(url, params=params, timeout=10)
    resp.raise_for_status()
    data = resp.json()
    cw = data.get("current_weather", {})
    return {
        "temperature_c": cw.get("temperature"),
        "windspeed_kmh": cw.get("windspeed"),
        "winddirection_deg": cw.get("winddirection"),
        "time": cw.get("time"),
    }

calculate_tool = FunctionTool.from_defaults(fn=calculate, name="calculate")
weather_tool = FunctionTool.from_defaults(fn=get_weather, name="get_weather")

Run a quick sanity check:

python -c "
from tools import calculate_tool, weather_tool
print(calculate_tool('(3 + 4) * 2'))
print(weather_tool(37.7749, -122.4194))
"

Expected output (values will differ for weather):

{'result': 14}
{'temperature_c': 14.2, 'windspeed_kmh': 12.3, 'winddirection_deg': 280, 'time': '2024-06-15T14:00'}

Configure the OpenAI-compatible client

LlamaIndex’s OpenAIAgent accepts an llm parameter that implements the BaseLLM interface. The easiest path is to use OpenAI from llama_index.llms.openai pointed at the n4n.ai base URL. This gives you automatic fallback when a provider is rate-limited or degraded, per-token usage metering, and cache-control hint forwarding without any extra code.

# agent.py
import os
from dotenv import load_dotenv
from llama_index.llms.openai import OpenAI
from llama_index.core.agent import FunctionCallingAgentWorker
from llama_index.core.agent import AgentRunner
from tools import calculate_tool, weather_tool

load_dotenv()

llm = OpenAI(
    model="gpt-4o-mini",  # any of the 240+ models n4n.ai serves
    api_key=os.getenv("N4N_API_KEY"),
    api_base=os.getenv("N4N_BASE_URL"),
    temperature=0.1,
)

worker = FunctionCallingAgentWorker.from_tools(
    [calculate_tool, weather_tool],
    llm=llm,
    verbose=True,
    system_prompt=(
        "You are a helpful assistant with access to a calculator and a weather service. "
        "Use the calculator for any arithmetic. Use the weather tool for current conditions. "
        "Always call tools when users ask for computations or live data."
    ),
)

agent = AgentRunner(worker)

The verbose=True flag prints each tool call and its result to stdout — useful for debugging the agent loop.

Run a single-turn query

# run_single.py
from agent import agent

response = agent.chat("What is (12 * 7) + 5, and what is the weather in San Francisco?")
print(response)

Execute it:

python run_single.py

Expected console trace (abridged):

=== Calling Function ===
Calling function: calculate with args: {"expression": "(12 * 7) + 5"}
Got output: {"result": 89}
========================
=== Calling Function ===
Calling function: get_weather with args: {"latitude": 37.7749, "longitude": -122.4194}
Got output: {"temperature_c": 14.2, "windspeed_kmh": 12.3, "winddirection_deg": 280, "time": "2024-06-15T14:00"}
========================
The calculation yields 89. Current weather in San Francisco: 14.2°C, wind 12.3 km/h from 280°.

The agent called both tools in sequence, composed the answer, and returned a natural-language response. Notice that you never specified which provider backs gpt-4o-mini — the gateway handles routing and fallback transparently.

Stream the response

Production interfaces benefit from token-by-token streaming. AgentRunner exposes astream_chat for async streaming and stream_chat for synchronous generators. Here is the synchronous version:

# run_stream.py
from agent import agent

def main():
    print("Streaming response:\n")
    for token in agent.stream_chat("Calculate 15% tip on $84.50 and tell me the weather in Tokyo."):
        print(token.delta, end="", flush=True)
    print()

if __name__ == "__main__":
    main()

Run it:

python run_stream.py

Output appears incrementally:

Streaming response:

The 15% tip on $84.50 is $12.68. Current weather in Tokyo: 22.1°C, wind 8.4 km/h from 190°.

Each token.delta is a string chunk. You can pipe this directly to a WebSocket, Server-Sent Events endpoint, or CLI progress bar.

Inspect per-token usage

The gateway returns usage metadata in the final response object. LlamaIndex surfaces it via response.raw when the underlying LLM client includes it. With the OpenAI-compatible client, usage arrives in the last chunk’s usage field.

# usage.py
from agent import agent

response = agent.chat("What is 2 ** 10?")
print("Response:", response)
print("Raw usage:", getattr(response, "raw", None))

Sample output:

Response: 2 ** 10 equals 1,024.
Raw usage: {'prompt_tokens': 187, 'completion_tokens': 18, 'total_tokens': 205}

You can aggregate these counters per session, per user, or per feature flag to drive cost attribution dashboards.

Handle multi-turn conversations

AgentRunner maintains conversation history automatically. Each chat or stream_chat call appends the user message and the assistant response to an internal ChatMemoryBuffer. The buffer respects the LLM’s context window and truncates older turns when necessary.

# multi_turn.py
from agent import agent

turns = [
    "My name is Alex.",
    "What is 42 * 2?",
    "Now add 100 to the previous result.",
    "What was my name again?",
]

for i, user_msg in enumerate(turns, 1):
    print(f"\n--- Turn {i} ---")
    print(f"User: {user_msg}")
    reply = agent.chat(user_msg)
    print(f"Agent: {reply}")

Output:

--- Turn 1 ---
User: My name is Alex.
Agent: Hello Alex! How can I help you today?

--- Turn 2 ---
User: What is 42 * 2?
Agent: 42 * 2 = 84.

--- Turn 3 ---
User: Now add 100 to the previous result.
Agent: 84 + 100 = 184.

--- Turn 4 ---
User: What was my name again?
Agent: Your name is Alex.

The agent recalled the name from turn 1 and the computed value from turn 2 without any explicit context management in your code.

Add a custom tool with structured output

Real systems often need tools that return Pydantic models for type safety downstream. LlamaIndex supports this via FunctionTool.from_defaults with a return_direct=False (default) so the agent sees the serialized JSON.

# structured_tool.py
from pydantic import BaseModel, Field
from llama_index.core.tools import FunctionTool
from typing import Optional
import httpx

class GeoResult(BaseModel):
    city: str
    country: str
    latitude: float
    longitude: float
    timezone: str

async def geocode(city: str) -> GeoResult:
    """Resolve a city name to coordinates using OpenStreetMap Nominatim."""
    async with httpx.AsyncClient(timeout=10.0) as client:
        resp = await client.get(
            "https://nominatim.openstreetmap.org/search",
            params={"q": city, "format": "json", "limit": 1},
            headers={"User-Agent": "llamaindex-openaiagent-tutorial/1.0"},
        )
        resp.raise_for_status()
        data = resp.json()
        if not data:
            raise ValueError(f"City not found: {city}")
        top = data[0]
        return GeoResult(
            city=city,
            country=top.get("display_name", "").split(",")[-1].strip(),
            latitude=float(top["lat"]),
            longitude=float(top["lon"]),
            timezone="UTC",  # Nominatim does not return tz; placeholder
        )

geocode_tool = FunctionTool.from_defaults(fn=geocode, name="geocode")

Register it with the agent by adding geocode_tool to the tool list in agent.py. The agent will now chain geocode → get_weather when a user asks “What’s the weather in Kyoto?” without you writing orchestration logic.

Error handling and retries

Network tools fail. The agent’s ReAct loop catches exceptions from tool calls and feeds the error message back to the LLM, which can decide to retry, switch tools, or apologize. You can also wrap tools with tenacity for automatic retries on transient errors:

# resilient_tool.py
from tenacity import retry, stop_after_attempt, wait_exponential_jitter
import httpx

@retry(
    wait=wait_exponential_jitter(initial=0.5, max=4),
    stop=stop_after_attempt(3),
)
async def robust_get(url: str, params: dict) -> httpx.Response:
    async with httpx.AsyncClient(timeout=10.0) as client:
        return await client.get(url, params=params)

Apply the decorator to get_weather or geocode to survive brief provider blips without surfacing raw stack traces to the user.

Deploying behind a gateway

When you move this to production, two operational details matter:

Cache-control hints — The gateway forwards Cache-Control headers from upstream providers. If your workload includes repeated identical prompts (e.g., classification over a fixed taxonomy), enable prompt caching on the provider side and the gateway will honor Cache-Control: public, max-age=... responses, reducing latency and cost.
Routing directives — Clients can pass X-Route: provider=anthropic or X-Route: priority=latency headers to steer traffic. In LlamaIndex, set default_headers on the OpenAI client:

llm = OpenAI(
    model="gpt-4o-mini",
    api_key=os.getenv("N4N_API_KEY"),
    api_base=os.getenv("N4N_BASE_URL"),
    default_headers={"X-Route": "priority=latency"},
)

This keeps routing policy in infrastructure config, not application code.

Summary

You now have a complete LlamaIndex OpenAIAgent that:

Calls typed tools (calculator, weather, geocoding) with automatic JSON serialization
Streams tokens for responsive UIs
Reports per-token usage for cost tracking
Maintains multi-turn memory without manual context management
Routes through a single OpenAI-compatible endpoint that addresses 240-plus models and falls back when a provider is rate-limited or degraded

The only n4n.ai-specific lines in the entire codebase are the api_base and api_key passed to the OpenAI client. Swap the base URL to another OpenAI-compatible gateway and the agent keeps working — no vendor lock-in, no bespoke SDKs.