Mocking the n4n API in your agent test suite

Mocking n4n API for agent tests is the fastest way to get deterministic, cost-free coverage of your LLM agent’s decision loops. This tutorial builds a pytest harness that intercepts the OpenAI-compatible endpoint and returns canned tool calls, streaming tokens, and fallback responses so you can test agent logic without burning tokens or waiting on live providers.

Prerequisites

• Python 3.11+
• openai>=1.40 (OpenAI-compatible client)
• respx and pytest for HTTP interception
• A minimal agent loop that calls client.chat.completions.create

Install the test deps:

pip install "openai>=1.40" respx pytest

We will point the OpenAI client at the n4n API endpoint but intercept every request with respx, so no traffic leaves your machine.

1. Configure the client

The n4n API is OpenAI-compatible, so the standard SDK works with a custom base_url.

# client_setup.py
import os
from openai import OpenAI

def get_client() -> OpenAI:
    return OpenAI(
        base_url="https://api.n4n.ai/v1",
        api_key=os.environ.get("N4N_API_KEY", "test-key"),
    )

In tests we never hit that URL; respx patches the transport layer.

2. Mock a tool-call response

A typical agent step expects the model to emit tool_calls. Mock the POST to /chat/completions with a fixed JSON body that matches the OpenAI schema.

# test_agent.py
import json
import respx
from httpx import Response
from client_setup import get_client

TOOL_CALL_RESP = {
    "id": "chatcmpl-1",
    "object": "chat.completion",
    "model": "gpt-4o-mini",
    "choices": [{
        "index": 0,
        "message": {
            "role": "assistant",
            "content": None,
            "tool_calls": [{
                "id": "call_1",
                "type": "function",
                "function": {
                    "name": "get_weather",
                    "arguments": json.dumps({"location": "San Francisco"})
                }
            }]
        },
        "finish_reason": "tool_calls"
    }],
    "usage": {"prompt_tokens": 12, "completion_tokens": 8, "total_tokens": 20}
}

Register the route:

@respx.mock
def test_first_tool_call():
    respx.post("https://api.n4n.ai/v1/chat/completions").mock(
        return_value=Response(200, json=TOOL_CALL_RESP)
    )
    client = get_client()
    resp = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Weather in SF?"}],
        tools=[{"type": "function", "function": {"name": "get_weather"}}],
    )
    assert resp.choices[0].message.tool_calls[0].function.name == "get_weather"

Expected output when run:

PASSED test_agent.py::test_first_tool_call

3. Test the full agent loop

A real agent processes the tool call, appends the result, and calls the model again. Mock both responses in sequence.

FINAL_RESP = {
    "id": "chatcmpl-2",
    "object": "chat.completion",
    "model": "gpt-4o-mini",
    "choices": [{
        "index": 0,
        "message": {"role": "assistant", "content": "It's foggy and 60°F in San Francisco."},
        "finish_reason": "stop"
    }],
    "usage": {"prompt_tokens": 30, "completion_tokens": 10, "total_tokens": 40}
}

def run_agent(client, user_msg):
    messages = [{"role": "user", "content": user_msg}]
    tools = [{
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Get weather",
            "parameters": {"type": "object", "properties": {"location": {"type": "string"}}}
        }
    }]
    resp = client.chat.completions.create(model="gpt-4o-mini", messages=messages, tools=tools)
    msg = resp.choices[0].message
    if msg.tool_calls:
        for tc in msg.tool_calls:
            args = json.loads(tc.function.arguments)
            # local stub for the tool
            result = {"temp": 60, "cond": "foggy"}
            messages.append(msg.model_dump())
            messages.append({"role": "tool", "tool_call_id": tc.id, "content": json.dumps(result)})
        resp2 = client.chat.completions.create(model="gpt-4o-mini", messages=messages, tools=tools)
        return resp2.choices[0].message.content
    return msg.content

@respx.mock
def test_agent_loop():
    url = "https://api.n4n.ai/v1/chat/completions"
    respx.post(url).mock(side_effect=[
        Response(200, json=TOOL_CALL_RESP),
        Response(200, json=FINAL_RESP),
    ])
    out = run_agent(get_client(), "Weather in SF?")
    assert out == "It's foggy and 60°F in San Francisco."

The sequence mock ensures the first call returns the tool request and the second returns the synthesized answer. Run with pytest -q and you should see a green check.

4. Mock streaming and per-token usage

Agents that stream need to parse Server-Sent Events. The n4n API returns text/event-stream with the same shape as OpenAI. Mock a streaming endpoint by returning raw SSE text.

def test_streaming(monkeypatch):
    import httpx
    @respx.mock
    def _run():
        def sse_handler(request):
            chunks = [
                'data: {"choices":[{"delta":{"content":"Hello"}}]}\n\n',
                'data: {"choices":[{"delta":{"content":" world"}}]}\n\n',
                'data: [DONE]\n\n',
            ]
            return Response(200, text="".join(chunks),
                            headers={"content-type": "text/event-stream"})
        respx.post("https://api.n4n.ai/v1/chat/completions").mock(side_effect=sse_handler)
        client = get_client()
        collected = []
        stream = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": "hi"}],
            stream=True,
        )
        for chunk in stream:
            if chunk.choices[0].delta.content:
                collected.append(chunk.choices[0].delta.content)
        assert "".join(collected) == "Hello world"
    _run()

Per-token metering appears in the non-streaming usage field; your mock should include it so agent code that logs cost doesn’t crash. The earlier TOOL_CALL_RESP already carries usage.

5. Simulate provider fallback

n4n.ai provides automatic fallback when a provider is rate-limited or degraded, so your mock should reflect that by returning a different model identifier in the response body. From the client’s perspective the request succeeds; only the model field changes.

FALLBACK_RESP = {
    "id": "chatcmpl-3",
    "object": "chat.completion",
    "model": "mistral-7b-instruct",  # gateway routed to a healthy provider
    "choices": [{
        "index": 0,
        "message": {"role": "assistant", "content": "Fallback response"},
        "finish_reason": "stop"
    }],
    "usage": {"prompt_tokens": 5, "completion_tokens": 3, "total_tokens": 8}
}

@respx.mock
def test_fallback_model_name():
    respx.post("https://api.n4n.ai/v1/chat/completions").mock(
        return_value=Response(200, json=FALLBACK_RESP)
    )
    client = get_client()
    resp = client.chat.completions.create(model="gpt-4o-mini", messages=[])
    assert resp.model == "mistral-7b-instruct"

This asserts your agent doesn’t hardcode model assumptions and can log the actual serving model for observability.

6. Run the suite and verify

Put the tests in test_agent.py and execute:

pytest -q

Expected output:

....                                                                  [100%]
4 passed in 0.31s

You now have deterministic coverage for tool calling, streaming, usage metering, and fallback routing. Because every response is intercepted, tests run in milliseconds and cost zero tokens.

Tips for maintainability

• Centralize response fixtures in a fixtures.py module so multiple tests share schema-valid bodies.
• If your agent passes provider cache-control hints via extra_headers, assert those headers arrive at the mock using respx.calls.last.request.headers.
• For complex multi-turn loops, use respx.post(url).mock(side_effect=response_list) and keep the list in the test name’s scenario.
• Treat the mocked schema as a contract: when the n4n API bumps the OpenAI compatibility version, update fixtures alongside the client bump.

Mocking the n4n API for agent tests turns flaky LLM integrations into ordinary unit tests. The pattern scales to hundreds of agents without touching a live endpoint.