Testing Agents That Search, Without Hitting the API

Recorded fixtures, deterministic replay, and the failure cases you should be simulating but probably aren't.

Profile picture of Serply
Serply
A test suite replaying recorded search responses

Agent tests that call a live search API are slow, flaky, and expensive, and they fail when someone’s ranking changes. Agent tests with hand-written mocks pass reliably and tell you nothing, because the mock returns a tidy shape the real API never produces.

The middle path is recording real responses once and replaying them.

Recording

import os
import json
import hashlib
from pathlib import Path
import requests
from urllib.parse import quote_plus

API_KEY = os.environ.get("SERPLY_API_KEY")
FIXTURES = Path(__file__).parent / "fixtures"
FIXTURES.mkdir(exist_ok=True)


def _key(kind: str, ident: str) -> Path:
    digest = hashlib.sha256(ident.encode()).hexdigest()[:16]
    return FIXTURES / f"{kind}_{digest}.json"


def record_search(query: str, num: int = 10) -> dict:
    path = _key("search", f"{query}|{num}")
    resp = requests.get(
        f"https://api.serply.io/v1/search/q={quote_plus(query)}&num={num}",
        headers={"X-Api-Key": API_KEY},
        timeout=30,
    )
    resp.raise_for_status()
    payload = {"query": query, "num": num, "status": resp.status_code, "body": resp.json()}
    path.write_text(json.dumps(payload, indent=2))
    return payload["body"]


def record_page(url: str) -> str:
    path = _key("page", url)
    resp = requests.post(
        "https://api.serply.io/v1/request",
        headers={"X-Api-Key": API_KEY, "Content-Type": "application/json"},
        json={"url": url, "response_type": "markdown"},
        timeout=60,
    )
    resp.raise_for_status()
    path.write_text(json.dumps({"url": url, "status": resp.status_code, "text": resp.text}))
    return resp.text

Store the whole body, not a trimmed version. The fields you don’t use today — ts, device_region, total, the empty ads and related_questions arrays — are exactly what will break your parser when you start using them.

Note the two different response formats: search returns JSON, while the scraper’s markdown mode returns text as the body. Record them differently or your replay will lie to you about the shape.

Replaying

class ReplayClient:
    def __init__(self, strict: bool = True):
        self.strict = strict
        self.calls: list[tuple[str, str]] = []

    def search(self, query: str, num: int = 10) -> dict:
        self.calls.append(("search", query))
        path = _key("search", f"{query}|{num}")
        if not path.exists():
            if self.strict:
                raise AssertionError(
                    f"No fixture for search({query!r}, num={num}). "
                    f"Run the recorder or add one at {path.name}."
                )
            return {"results": []}
        return json.loads(path.read_text())["body"]

    def read(self, url: str) -> str:
        self.calls.append(("read", url))
        path = _key("page", url)
        if not path.exists():
            if self.strict:
                raise AssertionError(f"No fixture for page {url}")
            return ""
        return json.loads(path.read_text())["text"]

Strict mode is worth defaulting to. A replay client that silently returns empty results for an unrecorded query turns a missing fixture into a mysterious behavioral change, and you’ll spend an afternoon on it.

Recording self.calls gives you assertions about behavior, which is usually what you care about:

def test_agent_searches_before_answering(replay):
    agent = build_agent(client=replay)
    agent.run("What's the current pricing for the Pro plan?")
    assert any(kind == "search" for kind, _ in replay.calls)


def test_agent_does_not_search_for_arithmetic(replay):
    agent = build_agent(client=replay)
    agent.run("What is 15% of 240?")
    assert not replay.calls

The cases you’re not testing

Happy-path fixtures are the easy half. These are the ones that actually break agents in production:

EMPTY_SEARCH = {
    "results": [], "ads": [], "ads_count": 0, "answers": [],
    "image_results": [], "shopping_ads": [], "places": [],
    "related_searches": [], "carousel": [], "company": {},
    "total": None, "knowledge_graph": "", "related_questions": [],
    "carousel_count": 0, "ts": 0.4, "device_region": "",
    "device_type": None, "query": "asdkjhasdkjh",
}

MISSING_DESCRIPTIONS = {
    "results": [
        {"title": "Some Page", "link": "https://example.com/a",
         "position": 1, "realPosition": 1, "result_type": "organic",
         "metadata": {}},
    ],
    "query": "test", "ts": 0.9, "total": None,
}

TRACKING_LINKS = {
    "results": [
        {"title": "Result", "position": 1, "realPosition": 1,
         "result_type": "organic", "metadata": {"display_url": "example.com"},
         "link": "https://www.google.com/url?client=internal&ved=2ahUK&usg=AOv"},
    ],
    "query": "test", "ts": 1.1, "total": None,
}

That last one is real and catches people out. Results served from certain buckets carry Google’s own tracking parameters rather than a clean destination URL. If your agent’s citation logic assumes link is always the publisher’s URL, this fixture is how you find out.

MISSING_DESCRIPTIONS covers the other common crash: description absent rather than empty-string. Code that does r["description"][:200] dies here, and it will happen in production within a week.

Failure injection

import pytest


class FailingClient(ReplayClient):
    def __init__(self, mode: str):
        super().__init__(strict=False)
        self.mode = mode

    def search(self, query: str, num: int = 10) -> dict:
        self.calls.append(("search", query))
        if self.mode == "rate_limit":
            raise requests.HTTPError(response=_response(429))
        if self.mode == "timeout":
            raise requests.Timeout("timed out")
        if self.mode == "empty":
            return EMPTY_SEARCH
        return super().search(query, num)


@pytest.mark.parametrize("mode", ["rate_limit", "timeout", "empty"])
def test_agent_degrades_gracefully(mode):
    agent = build_agent(client=FailingClient(mode))
    out = agent.run("What's the latest version of Python?")
    assert out                                  # it answered something
    assert len(FailingClient(mode).calls) < 10  # it didn't retry forever

The retry-storm assertion is the valuable one. An agent that treats 429 as “try a different query” will make thirty calls before giving up, and you will not notice until the bill arrives.

Keeping fixtures honest

Recorded fixtures rot. Run a nightly job that re-records and diffs the shape — not the content, which legitimately changes:

def shape(obj, depth=0):
    if depth > 3:
        return "..."
    if isinstance(obj, dict):
        return {k: shape(v, depth + 1) for k, v in sorted(obj.items())}
    if isinstance(obj, list):
        return [shape(obj[0], depth + 1)] if obj else []
    return type(obj).__name__

Compare shape(recorded) against shape(fresh). Rankings change daily and that’s noise; a field changing from str to null, or disappearing, is the signal — and it’s the thing that quietly breaks a parser you haven’t touched in months.