Giving an agent a web search tool that returns analyzed results

An agent with a search tool that returns ten titles and ten snippets has to spend a second inference pass deciding which links are worth fetching, then a third fetching them. Each hop adds latency and a chance to pick wrong.

POST /search collapses that: multi-source retrieval plus content analysis in one asynchronous job. It returns 202 with a job identifier, so the tool wrapper needs to poll before it can answer.

The enqueue call

import os
import requests

API = "https://api.messora.dev"
HEADERS = {"X-API-Key": os.environ["MESSORA_API_KEY"]}


def start_search(query: str, num_results: int = 10) -> tuple[str, int]:
    resp = requests.post(
        f"{API}/search",
        headers=HEADERS,
        json={
            "query": query,
            "num_results": num_results,
            "query_fanout": True,
        },
        timeout=30,
    )
    resp.raise_for_status()          # 202
    body = resp.json()
    return body["job_id"], body["credits_estimated"]

credits_estimated comes back with the enqueue, before any work happens. Billing is 1 credit per result, so a num_results of 10 reserves 10. Reading that number and logging it is how you keep an autonomous agent from quietly spending a month of balance in an afternoon.

Narrowing the search

Four filters matter more than the query wording:

json={
    "query": "postgres logical replication lag monitoring",
    "num_results": 8,
    "include_domains": ["postgresql.org", "aws.amazon.com"],
    "exclude_domains": ["pinterest.com", "quora.com"],
    "freshness": "month",
    "country": "BR",
}

include_domains is the strongest signal available. For technical queries, restricting to official documentation domains does more for answer quality than any amount of prompt engineering downstream, because it removes the aggregator pages that repeat each other.

query_fanout defaults to True and expands one query into several related ones before retrieval. Turn it off when you need the literal query and nothing adjacent — checking whether a specific error string appears anywhere, for example.

Wrapping it as a tool

import time

TERMINAL = {"SUCCESS", "FAILURE", "REVOKED"}


def web_search(query: str, num_results: int = 8, timeout_s: int = 180) -> list[dict]:
    """Tool function: returns analyzed search results or raises."""
    job_id, estimated = start_search(query, num_results)

    deadline = time.monotonic() + timeout_s
    delay = 2.0
    while time.monotonic() < deadline:
        resp = requests.get(f"{API}/jobs/{job_id}", headers=HEADERS, timeout=30)
        if resp.status_code == 429:
            time.sleep(delay)
            delay = min(delay * 2, 20.0)
            continue
        resp.raise_for_status()
        job = resp.json()
        if job["status"] in TERMINAL:
            if job["status"] != "SUCCESS":
                raise RuntimeError(job.get("error") or job["status"])
            return job["results"] or []
        time.sleep(delay)
        delay = min(delay * 1.5, 10.0)

    raise TimeoutError(f"search job {job_id} exceeded {timeout_s}s")

The tool docstring an agent reads should state the credit cost. A model that knows a call costs 8 credits behaves differently from one that thinks search is free — it batches questions instead of firing one per thought.

Rate limit shapes the orchestration

/search allows 2 enqueues per minute per account, the tightest limit on the public API. That is deliberate: each call fans out to multiple sources and runs analysis, so it is the most expensive operation available.

For an agent loop this means search cannot be the default action. Two patterns that survive the limit:

1. Search once, scrape many. One /search to find the right five domains, then /batch at 1 credit per page to read them.
2. Cache by normalized query. Lowercase, strip punctuation, sort tokens, and key a local dict on the result. Agents re-ask the same question with different phrasing constantly.

Budget guard worth adding

class SearchBudget:
    def __init__(self, max_credits: int):
        self.remaining = max_credits

    def spend(self, credits: int) -> None:
        if credits > self.remaining:
            raise RuntimeError(
                f"search budget exhausted: needs {credits}, has {self.remaining}"
            )
        self.remaining -= credits

Call spend(credits_estimated) right after the enqueue returns. A 402 from the API means the entire account balance is gone, which is a much worse place to discover the problem than a per-session ceiling you set yourself.

What the results carry

Job results arrive under results once the status reaches SUCCESS, with per-item content already analyzed rather than raw snippets. Feed them to the model directly; the second retrieval pass most agent frameworks add after a search call is unnecessary here, and each skipped pass is one less place to lose the thread.