Most agent stacks assume tools are reliable until they aren’t. When you’re testing agent retry logic with flaky tool APIs, you need a controlled way to inject failures without standing up a fragile external service. Unlike LLM inference gateways such as n4n.ai that abstract provider degradation with automatic fallback, your own tool APIs expose raw failure modes that will surface in production.
Why deterministic flakiness matters
Random network blips are impossible to assert against. You want a knob that says “fail the first three calls with 503, then succeed”. That turns retry logic from hope into a unit test.
A mock that always works gives false confidence. A mock that always fails breaks the happy path. The middle ground is a stateful fault injector that lets you script exactly which calls break and how they break.
Step 1: Stand up a configurable flaky HTTP server
Use a tiny FastAPI app. It reads failure configuration from environment variables and tracks call count in process memory. For a single-test process this is enough; if you run the server as a separate container, swap the dict for Redis or a file.
# flaky_tool.py
import os
import random
import asyncio
from fastapi import FastAPI, JSONResponse
app = FastAPI()
FAILURE_RATE = float(os.getenv("FAILURE_RATE", "0.0"))
LATENCY_MS = int(os.getenv("LATENCY_MS", "0"))
ERROR_CODE = int(os.getenv("ERROR_CODE", "503"))
FAIL_EVERY_N = int(os.getenv("FAIL_EVERY_N", "0"))
FAIL_FIRST_N = int(os.getenv("FAIL_FIRST_N", "0"))
TOTAL_CALLS = {"n": 0}
@app.get("/tool")
async def tool():
TOTAL_CALLS["n"] += 1
call_n = TOTAL_CALLS["n"]
if LATENCY_MS > 0:
await asyncio.sleep(LATENCY_MS / 1000.0)
if FAIL_FIRST_N and call_n <= FAIL_FIRST_N:
return JSONResponse(status_code=ERROR_CODE, content={"error": "forced early failure"})
if FAIL_EVERY_N and call_n % FAIL_EVERY_N == 0:
return JSONResponse(status_code=ERROR_CODE, content={"error": "periodic failure"})
if random.random() < FAILURE_RATE:
return JSONResponse(status_code=ERROR_CODE, content={"error": "random flake"})
return {"result": "ok", "call": call_n}Run it:
pip install fastapi uvicorn
FAIL_FIRST_N=2 LATENCY_MS=20 ERROR_CODE=503 uvicorn flaky_tool:app --port 8080This gives you a tool endpoint that throws a configurable fraction of 503s, fails the first N calls, or fails every Nth call. For testing agent retry logic with flaky tool APIs, the deterministic FAIL_FIRST_N and FAIL_EVERY_N modes are what make assertions meaningful.
Step 2: Define a retry policy in your tool client
Agents call tools through a client. Wrap that client with explicit backoff and retry limits. Don’t hide retries inside a generic HTTP library; you want them observable and bounded.
# client.py
import time
import requests
class FlakyToolClient:
def __init__(self, base_url, max_retries=5, backoff=0.1):
self.base_url = base_url
self.max_retries = max_retries
self.backoff = backoff
self.attempts = 0
def call(self):
self.attempts = 0
last_exc = None
for attempt in range(self.max_retries + 1):
self.attempts += 1
try:
r = requests.get(f"{self.base_url}/tool", timeout=2)
if r.status_code >= 500:
raise RuntimeError(f"status {r.status_code}")
return r.json()
except Exception as e:
last_exc = e
if attempt < self.max_retries:
time.sleep(self.backoff * (2 ** attempt))
raise RuntimeError(f"tool failed after {self.attempts} attempts: {last_exc}")The exponential backoff is minimal but real. In production you’d add jitter; for tests, determinism wins. Critically, only retry on idempotent operations—if your tool mutates state, blind retries will double-charge customers. Gate retries behind an idempotency_key or restrict them to GET-style reads.
Step 3: Wire the flaky tool into an agent loop
A minimal agent loops: decide action, call tool, observe. Here’s a stripped-down version that uses the client directly.
# agent.py
from client import FlakyToolClient
def run_agent(tool_client):
# Stub planning step: agent always decides to call the tool
obs = tool_client.call()
return obs["result"]If you’re using an LLM-driven agent, swap the planning stub for a completion call. The retry boundary stays at the tool client, not inside the model call. The model should see a clean observation: either the tool result or a structured “tool unavailable” message after retries exhaust.
Step 4: Drive failure scenarios from tests
pytest with FastAPI’s TestClient avoids port juggling. For an integration test that mirrors production, start the uvicorn server in a fixture and point the client at it.
# test_retries.py
import pytest
import subprocess
import time
import requests
from client import FlakyToolClient
@pytest.fixture
def flaky_server():
env = {"FAIL_FIRST_N": "2", "ERROR_CODE": "503", "LATENCY_MS": "10"}
proc = subprocess.Popen(
["uvicorn", "flaky_tool:app", "--port", "8099"],
env={**__import__("os").environ, **env}
)
time.sleep(2) # wait for boot
yield "http://localhost:8099"
proc.terminate()
def test_agent_recovers_from_two_failures(flaky_server):
client = FlakyToolClient(flaky_server, max_retries=5, backoff=0.05)
result = None
for _ in range(3): # agent loop retries
try:
result = run_agent(client)
break
except RuntimeError:
pass
assert result == "ok"
assert client.attempts == 3 # two failures + one successThis asserts both recovery and exact retry count. When testing agent retry logic with flaky tool APIs, the attempts == 3 line is the real proof—not just a green test.
Step 5: Measure and assert retry behavior under load
One call is easy. Ten concurrent agents is where connection pools exhaust. Extend the client with a simple metric emitter.
import logging
class InstrumentedToolClient(FlakyToolClient):
def call(self):
super().call()
logging.info("tool_attempts", extra={"attempts": self.attempts})
return self.last_result if hasattr(self, "last_result") else NoneIn the base call, store self.last_result = r.json() before returning. Now a load test with pytest-xdist can confirm that under FAILURE_RATE=0.5, 95% of agents still finish within max_retries. You don’t need exact stats—just confirm the histogram of attempts centers where your fault config says it should.
Step 6: Simulate latency-induced timeouts
Flakiness isn’t only status codes. A tool that hangs breaks agents differently. Set LATENCY_MS=2000 and give the client timeout=0.5. The retry should trigger on requests.exceptions.Timeout.
def test_timeout_retry(flaky_server_with_latency):
client = FlakyToolClient(flaky_server_with_latency, max_retries=3, backoff=0.01)
try:
client.call()
except RuntimeError:
pass
assert client.attempts > 1This exercises the timeout branch, which is where many agent frameworks silently deadlock because they treat a hung socket as a successful stream. If your agent uses an event loop, use httpx.AsyncClient with timeout and asyncio.wait_for to catch coroutine stalls.
Step 7: Parameterize scenarios and clean up
Don’t write one test per failure mode. Drive them from a table.
@pytest.mark.parametrize("env,expected_attempts", [
({"FAIL_FIRST_N": "2"}, 3),
({"FAIL_EVERY_N": "2"}, 2),
({"LATENCY_MS": "2000"}, 4), # assuming timeout=0.1, backoff small
])
def test_scenarios(env, expected_attempts):
# start server with env, run agent, assert attempts
...After the suite, kill any subprocesses and clear TOTAL_CALLS if you import the module directly. Leftover state is the silent killer of flaky tests about flaky tools.
Verify success
Run the suite:
pytest test_retries.py -vYou should see green tests where the agent recovered from injected 503s and timeouts, and the attempts counter matches your fault schedule. If you bump FAIL_FIRST_N beyond max_retries, the test should fail with RuntimeError—that’s the expected red signal proving the retry cap works.
For end-to-end confidence, start the uvicorn server with FAIL_FIRST_N=4 and point a real agent at http://localhost:8080. Watch logs: you want to see the agent emit a tool call, receive an error, wait, and call again. When call five returns 200, the agent proceeds.
Testing agent retry logic with flaky tool APIs is not glamorous, but it’s the difference between a demo that works on your laptop and a system that survives a vendor outage. Build the knob, wire the counter, assert the schedule.