Tool calling in LangChain with GPT-4o via n4n.ai

LangChain tool calling with GPT-4o via n4n.ai gives you a single OpenAI-compatible endpoint that routes across 240+ models while handling provider fallbacks and usage metering automatically. This tutorial walks through the complete setup: configuring the client, defining tools with Pydantic schemas, binding them to the model, and executing calls in a loop that handles both tool invocations and final responses. You’ll end up with a minimal, production-ready pattern you can extend into agents or multi-step workflows.

Step 1: install dependencies

Start with a clean virtual environment. You need LangChain’s OpenAI integration, the core library, and Pydantic for tool schemas.

python -m venv .venv
source .venv/bin/activate
pip install "langchain-openai>=0.2.0" "langchain-core>=0.3.0" pydantic python-dotenv

Verify the imports work:

# test_imports.py
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage
from pydantic import BaseModel, Field
print("imports ok")

Run it:

python test_imports.py

Step 2: configure the n4n.ai client

n4n.ai exposes an OpenAI-compatible base URL. Point the LangChain ChatOpenAI client at it and pass your API key. The client honors standard OpenAI parameters plus any provider-specific hints forwarded via headers.

# config.py
import os
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()

N4N_API_KEY = os.getenv("N4N_API_KEY")
N4N_BASE_URL = "https://api.n4n.ai/v1"  # OpenAI-compatible endpoint

if not N4N_API_KEY:
    raise RuntimeError("Set N4N_API_KEY in .env or environment")

def get_llm(model: str = "gpt-4o", temperature: float = 0) -> ChatOpenAI:
    return ChatOpenAI(
        model=model,
        temperature=temperature,
        api_key=N4N_API_KEY,
        base_url=N4N_BASE_URL,
        # Optional: request specific provider or enable cache hints
        # default_headers={"x-n4n-provider": "openai", "x-n4n-cache": "true"},
    )

Create a .env file:

# .env
N4N_API_KEY=your_n4n_api_key_here

Test the connection with a simple completion:

# test_connection.py
from config import get_llm

llm = get_llm()
resp = llm.invoke([HumanMessage(content="Say 'pong' if you receive this")])
print(resp.content)

Expected output: pong (or similar acknowledgment).

Step 3: define tools with Pydantic schemas

LangChain’s @tool decorator accepts a Pydantic model for the argument schema. This gives you validation, documentation, and automatic JSON Schema generation for the model. Define two tools: a calculator and a weather lookup (stubbed for demonstration).

# tools.py
from langchain_core.tools import tool
from pydantic import BaseModel, Field

class CalculatorInput(BaseModel):
    expression: str = Field(description="A valid Python arithmetic expression, e.g., '(2 + 3) * 4'")

@tool(args_schema=CalculatorInput)
def calculator(expression: str) -> str:
    """Evaluate a basic arithmetic expression safely."""
    allowed_names = {"abs": abs, "round": round, "min": min, "max": max, "pow": pow}
    try:
        result = eval(expression, {"__builtins__": {}}, allowed_names)
        return str(result)
    except Exception as e:
        return f"Error: {e}"

class WeatherInput(BaseModel):
    location: str = Field(description="City and optional state/country, e.g., 'San Francisco, CA'")
    units: str = Field(default="fahrenheit", description="fahrenheit or celsius")

@tool(args_schema=WeatherInput)
def get_weather(location: str, units: str = "fahrenheit") -> str:
    """Stub weather tool — replace with a real API call in production."""
    # In a real implementation, call OpenWeatherMap, WeatherAPI, etc.
    temp = 72 if units == "fahrenheit" else 22
    return f"Weather in {location}: {temp}°{'F' if units == 'fahrenheit' else 'C'}, partly cloudy"

Verify the schemas generate correctly:

# test_tools.py
from tools import calculator, get_weather
import json

print("calculator schema:")
print(json.dumps(calculator.args_schema.model_json_schema(), indent=2))
print("\nget_weather schema:")
print(json.dumps(get_weather.args_schema.model_json_schema(), indent=2))

Step 4: bind tools to the model

Use bind_tools to attach the tool definitions to the model. This injects the function schemas into the request so GPT-4o can decide when to call them. The model returns AIMessage objects with tool_calls populated when it wants to invoke a tool.

# agent.py
from config import get_llm
from tools import calculator, get_weather
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage

TOOLS = [calculator, get_weather]

def build_chain():
    llm = get_llm(model="gpt-4o", temperature=0)
    return llm.bind_tools(TOOLS)

def invoke_with_tools(chain, messages):
    """Single turn: invoke model, execute any tool calls, return updated message list."""
    response = chain.invoke(messages)
    messages.append(response)

    if not response.tool_calls:
        return messages

    # Execute each tool call sequentially
    for tool_call in response.tool_calls:
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]
        tool_id = tool_call["id"]

        # Dispatch to the correct function
        if tool_name == "calculator":
            result = calculator.invoke(tool_args)
        elif tool_name == "get_weather":
            result = get_weather.invoke(tool_args)
        else:
            result = f"Unknown tool: {tool_name}"

        # Append tool result as a ToolMessage
        messages.append(ToolMessage(content=str(result), tool_call_id=tool_id))

    # After tool results, invoke again for the final answer
    final_response = chain.invoke(messages)
    messages.append(final_response)
    return messages

Step 5: run a multi-turn conversation

Wire it together in a script that demonstrates a full loop: user asks a question requiring tools, model calls them, results are fed back, model produces a final answer.

# main.py
from agent import build_chain, invoke_with_tools
from langchain_core.messages import HumanMessage

def main():
    chain = build_chain()
    messages = [
        HumanMessage(content="What's 23 * 47, and what's the weather in Tokyo?")
    ]

    print("User:", messages[0].content)
    messages = invoke_with_tools(chain, messages)

    # The last message is the final AI response
    final = messages[-1]
    print("\nAssistant:", final.content)

    # Optional: print the full trace for debugging
    print("\n--- Full message trace ---")
    for i, m in enumerate(messages):
        print(f"[{i}] {m.type}: {getattr(m, 'content', '')[:200]}")
        if hasattr(m, "tool_calls") and m.tool_calls:
            print(f"    tool_calls: {m.tool_calls}")

if __name__ == "__main__":
    main()

Run it:

python main.py

Expected output (values may vary):

User: What's 23 * 47, and what's the weather in Tokyo?

Assistant: 23 * 47 = 1081. The weather in Tokyo is 72°F, partly cloudy.

--- Full message trace ---
[0] human: What's 23 * 47, and what's the weather in Tokyo?
[1] ai: 
    tool_calls: [{'name': 'calculator', 'args': {'expression': '23 * 47'}, 'id': 'call_abc123'}, {'name': 'get_weather', 'args': {'location': 'Tokyo'}, 'id': 'call_def456'}]
[2] tool: 1081
[3] tool: Weather in Tokyo: 72°F, partly cloudy
[4] ai: 23 * 47 = 1081. The weather in Tokyo is 72°F, partly cloudy.

Step 6: handle streaming and async

Production workloads often need streaming for latency-sensitive UIs. LangChain supports astream and astream_log for token-by-token output. The tool-call loop stays the same; you just stream the final response after tools complete.

# streaming.py
import asyncio
from agent import build_chain, invoke_with_tools
from langchain_core.messages import HumanMessage

async def main():
    chain = build_chain()
    messages = [HumanMessage(content="Calculate (15 + 27) * 3 and stream the answer")]

    # Run tool loop non-streaming (tools are fast)
    messages = invoke_with_tools(chain, messages)

    # Stream the final response
    print("Assistant (streaming): ", end="", flush=True)
    async for chunk in chain.astream(messages):
        if chunk.content:
            print(chunk.content, end="", flush=True)
    print()

if __name__ == "__main__":
    asyncio.run(main())

Step 7: add structured output for the final answer

If you need the final response in a specific schema (e.g., for downstream parsing), use with_structured_output on a separate model instance after the tool loop completes. Keep the tool-bound model for reasoning; use a second call for formatting.

# structured_output.py
from config import get_llm
from agent import build_chain, invoke_with_tools
from langchain_core.messages import HumanMessage
from pydantic import BaseModel, Field

class Answer(BaseModel):
    calculation_result: float = Field(description="Result of the arithmetic")
    weather_summary: str = Field(description="One-sentence weather description")
    confidence: float = Field(ge=0, le=1, description="Model confidence in the answer")

def main():
    chain = build_chain()
    messages = [HumanMessage(content="What's 12 * 13 and weather in London?")]

    messages = invoke_with_tools(chain, messages)

    # Extract tool results for the formatter
    tool_results = [m.content for m in messages if m.type == "tool"]
    context = "\n".join(tool_results)

    # Separate model for structured formatting
    formatter = get_llm(model="gpt-4o", temperature=0).with_structured_output(Answer)
    structured = formatter.invoke([
        HumanMessage(content=f"Format this data into the schema:\n{context}")
    ])

    print("Structured answer:")
    print(structured.model_dump_json(indent=2))

if __name__ == "__main__":
    main()

Output:

{
  "calculation_result": 156.0,
  "weather_summary": "London is 72°F and partly cloudy.",
  "confidence": 0.95
}

Step 8: error handling and retries

Tool calls can fail — network errors, rate limits, invalid arguments. Wrap the dispatch logic with retries and fallbacks. n4n.ai automatically retries across providers when a model is rate-limited or degraded, but your application should still handle tool-level failures gracefully.

# robust_agent.py
import time
from functools import wraps
from langchain_core.messages import ToolMessage
from tools import calculator, get_weather

def retry(max_attempts: int = 3, base_delay: float = 0.5):
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            last_exc = None
            for attempt in range(max_attempts):
                try:
                    return fn(*args, **kwargs)
                except Exception as e:
                    last_exc = e
                    if attempt < max_attempts - 1:
                        time.sleep(base_delay * (2 ** attempt))
            raise last_exc
        return wrapper
    return decorator

@retry(max_attempts=3)
def safe_calculator(args: dict) -> str:
    return calculator.invoke(args)

@retry(max_attempts=3)
def safe_get_weather(args: dict) -> str:
    return get_weather.invoke(args)

TOOL_DISPATCH = {
    "calculator": safe_calculator,
    "get_weather": safe_get_weather,
}

def invoke_with_tools_robust(chain, messages):
    response = chain.invoke(messages)
    messages.append(response)

    if not response.tool_calls:
        return messages

    for tool_call in response.tool_calls:
        tool_name = tool_call["name"]
        tool_args = tool_call["args"]
        tool_id = tool_call["id"]

        dispatch_fn = TOOL_DISPATCH.get(tool_name)
        if not dispatch_fn:
            result = f"Unknown tool: {tool_name}"
        else:
            try:
                result = dispatch_fn(tool_args)
            except Exception as e:
                result = f"Tool {tool_name} failed after retries: {e}"

        messages.append(ToolMessage(content=str(result), tool_call_id=tool_id))

    final_response = chain.invoke(messages)
    messages.append(final_response)
    return messages

Step 9: observability — log usage and latency

n4n.ai returns per-token usage in the response metadata and forwards provider cache-control hints. Capture these for cost tracking and debugging.

# observability.py
import time
from agent import build_chain, invoke_with_tools
from langchain_core.messages import HumanMessage

def run_with_metrics(chain, messages):
    start = time.perf_counter()
    messages = invoke_with_tools(chain, messages)
    elapsed = time.perf_counter() - start

    final = messages[-1]
    usage = getattr(final, "usage_metadata", None) or getattr(final, "response_metadata", {}).get("token_usage", {})

    print(f"Latency: {elapsed:.2f}s")
    print(f"Usage: {usage}")
    print(f"Response: {final.content}")
    return messages

def main():
    chain = build_chain()
    messages = [HumanMessage(content="What is 99 * 99?")]
    run_with_metrics(chain, messages)

if __name__ == "__main__":
    main()

Typical metadata includes input_tokens, output_tokens, total_tokens, and sometimes cache_read_tokens if the provider supports prompt caching.

Step 10: verification checklist

Before shipping, run through these checks:

1. Tool schemas match function signatures — run test_tools.py and confirm JSON Schema properties align with your Pydantic models.
2. Model actually calls tools — the trace in main.py should show tool_calls on the first AI message, not a direct answer.
3. Tool results feed back correctly — each ToolMessage must have a tool_call_id matching the originating call.
4. Final answer incorporates tool data — the last message should reference the computed values, not hallucinate.
5. Streaming works — streaming.py should print tokens incrementally after the tool loop.
6. Structured output validates — structured_output.py should produce JSON that passes Answer.model_validate().
7. Retries trigger — temporarily break a tool (raise an exception) and confirm robust_agent.py retries and eventually surfaces an error message rather than crashing.
8. Usage metadata present — observability.py should print non-empty token counts.

Common pitfalls

Extending from here

• Agents: Wrap the loop in langgraph.prebuilt.create_react_agent for multi-step reasoning with state.
• Parallel tool calls: GPT-4o supports multiple tool_calls in one response; the loop above already handles them sequentially — swap to asyncio.gather for parallel execution.
• Custom tool routing: Use n4n.ai’s x-n4n-provider header to pin specific tools to specific providers (e.g., code execution on a provider with a code interpreter).
• Caching: Enable x-n4n-cache: true on idempotent tool calls to reduce latency and cost on repeated queries.

You now have a complete, observable, retry-hardened tool-calling pipeline using LangChain, GPT-4o, and n4n.ai. The pattern scales from single-function helpers to multi-agent systems without changing the core loop.