One Question, Six Endpoints, One Round Trip

Some questions need web results, news, and local listings at once. Fanning out across endpoints costs the latency of the slowest one, not the sum.

Profile picture of Serply
Serply
Concurrent requests to several API endpoints merging into one result

“Is this restaurant chain still expanding?” touches three surfaces: news for announcements, maps for actual locations, web for analysis. An agent that calls them one at a time takes three round trips to learn what it could have learned in one.

The fix is unglamorous — concurrency — but doing it well means deciding which endpoints a question needs, handling partial failure, and merging results the model can actually reason over.

The clients

import os
import asyncio
import httpx
from urllib.parse import quote_plus

API_KEY = os.environ["SERPLY_API_KEY"]
BASE = "https://api.serply.io/v1"
HEADERS = {"X-Api-Key": API_KEY, "X-Proxy-Location": "US"}


async def web(c: httpx.AsyncClient, q: str, num: int = 10) -> dict:
    r = await c.get(f"{BASE}/search/q={quote_plus(q)}&num={num}",
                    headers=HEADERS, timeout=30)
    r.raise_for_status()
    return {"source": "web", "items": [
        {"title": i.get("title"), "url": i.get("link"),
         "text": i.get("description"), "rank": i.get("position")}
        for i in r.json().get("results", [])
    ]}


async def bing(c: httpx.AsyncClient, q: str, num: int = 10) -> dict:
    r = await c.get(f"{BASE}/b/search/q={quote_plus(q)}&num={num}",
                    headers=HEADERS, timeout=30)
    r.raise_for_status()
    return {"source": "bing", "items": [
        {"title": i.get("title"), "url": i.get("link"),
         "text": i.get("description"), "rank": i.get("position")}
        for i in r.json().get("results", [])
    ]}


async def news(c: httpx.AsyncClient, q: str) -> dict:
    r = await c.get(f"{BASE}/news/q={quote_plus(q)}", headers=HEADERS, timeout=30)
    r.raise_for_status()
    entries = r.json().get("feed", {}).get("entries", [])
    return {"source": "news", "items": [
        {"title": e.get("title"), "url": e.get("link"),
         "text": e.get("summary"), "published": e.get("published"),
         "publisher": e.get("source")}
        for e in entries
    ]}


async def maps(c: httpx.AsyncClient, q: str) -> dict:
    r = await c.get(f"{BASE}/maps/search/q={quote_plus(q)}",
                    headers=HEADERS, timeout=30)
    r.raise_for_status()
    return {"source": "maps", "items": [
        {"title": p.get("name"), "url": p.get("website") or p.get("google_maps_url"),
         "text": p.get("address"), "rating": p.get("rating"),
         "reviews": p.get("review_count"), "lat": p.get("latitude"),
         "lon": p.get("longitude")}
        for p in r.json().get("places", [])
    ]}


async def video(c: httpx.AsyncClient, q: str, num: int = 10) -> dict:
    r = await c.get(f"{BASE}/video/q={quote_plus(q)}&num={num}",
                    headers=HEADERS, timeout=30)
    r.raise_for_status()
    return {"source": "video", "items": [
        {"title": v.get("title"), "url": v.get("link"), "text": ""}
        for v in r.json().get("results", [])
    ]}


async def product(c: httpx.AsyncClient, q: str) -> dict:
    r = await c.get(f"{BASE}/product/search/q={quote_plus(q)}",
                    headers=HEADERS, timeout=30)
    r.raise_for_status()
    return {"source": "product", "items": [
        {"title": p.get("title"), "url": p.get("link"),
         "price": p.get("price"), "rating": p.get("rating_stars"),
         "reviews": p.get("review_count"), "sponsored": bool(p.get("is_sponsor"))}
        for p in r.json().get("products", [])
    ]}

Every endpoint gets normalised to {title, url, text} plus its own extras. That uniformity is what makes the merge step trivial later — and it papers over the real shape differences: web and video return results, news returns feed.entries, maps returns places, product returns products.

Choosing endpoints per question

Calling all six on every question is wasteful. Route first:

import re

SIGNALS = {
    "news": re.compile(
        r"\b(news|announced|announcement|latest|recent|this week|breaking|"
        r"report(ed|s)?|launch|acquisition|funding|earnings)\b", re.I),
    "maps": re.compile(
        r"\b(near me|nearby|location|locations|address|open now|hours|"
        r"restaurant|store|branch|office|directions|in [A-Z][a-z]+)\b"),
    "product": re.compile(
        r"\b(buy|price|cheapest|best .* (for|under)|review|amazon|"
        r"which .* should i (buy|get))\b", re.I),
    "video": re.compile(
        r"\b(tutorial|how to|demo|walkthrough|watch|video|review of)\b", re.I),
}


def choose(question: str) -> list[str]:
    picked = ["web"]
    for name, pattern in SIGNALS.items():
        if pattern.search(question):
            picked.append(name)
    return picked

Regex routing handles most traffic and costs nothing. For the ambiguous remainder, a small model classifying into the same set is a reasonable second pass — but measure before you add it, because the marginal accuracy is often not worth the latency you just added to the critical path.

The fan-out

DISPATCH = {"web": web, "bing": bing, "news": news,
            "maps": maps, "video": video, "product": product}


async def gather_sources(question: str, query: str | None = None) -> dict:
    query = query or question
    names = choose(question)

    async with httpx.AsyncClient() as c:
        tasks = [DISPATCH[n](c, query) for n in names]
        settled = await asyncio.gather(*tasks, return_exceptions=True)

    ok, failed = {}, {}
    for name, outcome in zip(names, settled):
        if isinstance(outcome, BaseException):
            failed[name] = f"{type(outcome).__name__}: {outcome}"
        else:
            ok[name] = outcome["items"]

    return {"query": query, "sources": ok, "failed": failed}

return_exceptions=True is the line that makes this production-safe. Without it, one endpoint timing out cancels the gather and you lose five successful responses along with the failed one.

Keeping failed rather than discarding it matters for the prompt — an answer built on four of five sources should be able to say which one was missing.

Bounding the slow ones

The fan-out takes as long as the slowest endpoint, which means one slow call sets your latency floor. Cap it:

async def gather_bounded(question: str, budget_seconds: float = 8.0) -> dict:
    query = question
    names = choose(question)

    async with httpx.AsyncClient() as c:
        tasks = {asyncio.create_task(DISPATCH[n](c, query)): n for n in names}
        done, pending = await asyncio.wait(tasks.keys(), timeout=budget_seconds)

        for t in pending:
            t.cancel()

        ok, failed = {}, {n: "timed out" for t, n in tasks.items() if t in pending}
        for t in done:
            name = tasks[t]
            try:
                ok[name] = t.result()["items"]
            except Exception as e:
                failed[name] = f"{type(e).__name__}: {e}"

    return {"query": query, "sources": ok, "failed": failed}

A hard wall-clock budget across the whole fan-out is more useful than per-request timeouts, because it’s the number your users actually experience.

Formatting for the model

Keep the surfaces labelled. Flattening everything into one list loses the information that made the fan-out worth doing:

def format_context(bundle: dict) -> str:
    labels = {
        "web": "WEB RESULTS (general, ranked by relevance)",
        "bing": "BING RESULTS (second engine — agreement with web results is a "
                "confidence signal)",
        "news": "NEWS ARTICLES (recent coverage, with publication dates)",
        "maps": "PLACES (real listings with addresses, ratings, coordinates)",
        "video": "VIDEOS (titles and links only — contents NOT retrieved)",
        "product": "PRODUCT LISTINGS (prices and ratings as displayed; some sponsored)",
    }
    parts = []
    for name, items in bundle["sources"].items():
        if not items:
            continue
        lines = [labels.get(name, name.upper())]
        for i in items[:10]:
            row = f"- {i.get('title')}"
            if i.get("url"):
                row += f"\n  {i['url']}"
            if i.get("published"):
                row += f"\n  published: {i['published']}"
            if i.get("price"):
                row += f"\n  price: {i['price']}"
            if i.get("rating"):
                row += f"\n  rating: {i['rating']} ({i.get('reviews')} reviews)"
            if i.get("text"):
                row += f"\n  {i['text']}"
            lines.append(row)
        parts.append("\n".join(lines))

    if bundle["failed"]:
        parts.append("UNAVAILABLE SOURCES: " + ", ".join(
            f"{k} ({v})" for k, v in bundle["failed"].items()
        ) + "\nDo not speculate about what these might have contained.")

    return "\n\n".join(parts)

The video label earns its place. Given a list of video titles with no other framing, models write summaries of videos they haven’t watched — stating the limitation inline, next to the data, is what prevents it.

The unavailable-sources note does the same job from the other direction: it tells the model a gap is a gap, not an absence of facts.

When not to do this

Fan-out is right when a question genuinely spans surfaces. It’s wrong as a default. Most questions need web results and nothing else, and calling six endpoints to answer “what does this error message mean” spends five requests to add noise to the context.

Route first, fan out second.