Search Agent: Mercury + Web Search

This cookbook demonstrates how to build an agentic web search loop — PLAN → SEARCH → ANALYZE → refine or FINISH → ANSWER with Mercury as the agent. Because each planning turn returns in a few hundred milliseconds, most of the wall clock goes to the search API rather than the model. We walk you through how to build:

A pluggable search function and an evidence store with stable citation IDs
A JSON action contract and the plan-search-refine loop it drives
A fully cited answer with a latency breakdown

Run the Search Agent

Directly via Google Colab or download the .ipynb to run it locally in Jupyter

Dependencies

%pip install -q "openai>=1.0" httpx

import asyncio, getpass, json, os, re, time

import httpx
from openai import AsyncOpenAI
from IPython.display import Markdown, display

API Keys

You need an Inception key plus a key for your search backend. Only the key for the backend you select is prompted for:

Key	Where
`INCEPTION_API_KEY`	platform.inceptionlabs.ai
`EXA_API_KEY`	dashboard.exa.ai — if `SEARCH_BACKEND = "exa"`
`PARALLEL_API_KEY`	docs.parallel.ai — if `SEARCH_BACKEND = "parallel"`

SEARCH_BACKEND = "exa"        # "exa" or "parallel"

BACKEND_KEYS = {"exa": "EXA_API_KEY", "parallel": "PARALLEL_API_KEY"}

for var in ("INCEPTION_API_KEY", BACKEND_KEYS[SEARCH_BACKEND]):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")

INCEPTION_BASE_URL = "https://api.inceptionlabs.ai/v1"   # OpenAI-compatible
MODEL = "mercury-2"

The Search Backend

The loop only ever sees {heading: [docs]}, where a doc is {url, title, highlights[], publishedDate}. Keeping that shape fixed means swapping providers — Brave, Tavily, Serper, or your own vector index over internal docs — is one function plus a branch in search().

def check(r: httpx.Response) -> httpx.Response:
    """raise_for_status, but keep the server's explanation of what it rejected."""
    if r.is_error:
        raise RuntimeError(f"{r.request.method} {r.request.url} -> {r.status_code}: {r.text[:400]}")
    return r


async def exa_search(http: httpx.AsyncClient, query: str, num_results: int) -> list[dict]:
    """Run one Exa /search with highlights, retrying on rate limits."""
    body = {
        "query": query,
        "type": "auto",
        "numResults": num_results,
        "contents": {"highlights": {"maxCharacters": 2000}},
    }
    headers = {"x-api-key": os.environ["EXA_API_KEY"], "Content-Type": "application/json"}
    for attempt in range(5):
        r = await http.post("https://api.exa.ai/search", json=body, headers=headers)
        if (r.status_code == 429 or r.status_code >= 500) and attempt < 4:
            await asyncio.sleep(2**attempt)          # bursty fanout will hit 429s
            continue
        return check(r).json().get("results", [])
    return []


async def search(http: httpx.AsyncClient, queries: list[str], *, objective: str,
                 num_results: int, backend: str) -> dict[str, list[dict]]:
    """One round of search -> {transcript heading: [docs]}. Add a provider here."""
    if backend == "exa":
        per_query = await asyncio.gather(*(exa_search(http, q, num_results) for q in queries))
        return dict(zip(queries, per_query))
    if backend == "parallel":
        return await parallel_search(http, queries, objective=objective, num_results=num_results)
    raise ValueError(f"unknown backend: {backend!r} (expected 'exa' or 'parallel')")

Optional: Parallel as the backend

async def parallel_search(http: httpx.AsyncClient, queries: list[str], *,
                          objective: str, num_results: int) -> dict[str, list[dict]]:
    body = {
        "objective": objective,
        "search_queries": queries,
        "mode": "turbo",
        "advanced_settings": {"max_results": num_results * len(queries)},
    }
    headers = {"x-api-key": os.environ["PARALLEL_API_KEY"], "Content-Type": "application/json"}
    for attempt in range(5):
        r = await http.post("https://api.parallel.ai/v1/search", json=body, headers=headers)
        if (r.status_code == 429 or r.status_code >= 500) and attempt < 4:
            await asyncio.sleep(2**attempt)
            continue
        return {" | ".join(queries): [            # one combined transcript heading
            {"url": it.get("url"), "title": it.get("title"),
             "highlights": it.get("excerpts") or [], "publishedDate": it.get("publish_date")}
            for it in check(r).json().get("results", [])
        ]}
    return {}

Evidence Bookkeeping

The agent’s memory. Two details are what make multi-round search work:

Stable source IDs. A URL keeps the same [n] for the whole run, across rounds. That is what lets the final answer cite [3] and have it mean something.
Append-only transcript. Each round is appended, never rewritten, so the prompt prefix stays byte-identical between rounds.

class Evidence:
    def __init__(self):
        self.blocks: list[str] = []      # one formatted block per search round
        self.ids: dict[str, int] = {}    # url -> stable source id
        self.titles: dict[str, str] = {}

    def add_round(self, round_no: int, results: dict[str, list[dict]]) -> int:
        """Append one search round. Returns the number of previously unseen sources."""
        new, sections = 0, []
        for query, docs in results.items():
            entries = []
            for doc in docs:
                url = (doc.get("url") or "").strip()
                if not url:
                    continue
                if url not in self.ids:                       # first sighting -> new id
                    self.ids[url] = len(self.ids) + 1
                    self.titles[url] = doc.get("title") or "(untitled)"
                    new += 1
                snippets = "\n".join(f"- {h.strip()}" for h in doc.get("highlights") or []) or "- (none)"
                date = f"\nDate: {doc['publishedDate']}" if doc.get("publishedDate") else ""
                entries.append(
                    f"#### Source [{self.ids[url]}]\nURL: {url}\n"
                    f"Title: {self.titles[url]}{date}\nHighlights:\n{snippets}"
                )
            sections.append(f"### Query: {query}\n\n" + "\n\n".join(entries))
        self.blocks.append(f"## Search round {round_no}\n\n" + "\n\n".join(sections))
        return new

    def transcript(self) -> str:
        return "\n\n".join(self.blocks) if self.blocks else "No searches have been run yet."

    def source_list(self) -> str:
        by_id = sorted(self.ids.items(), key=lambda kv: kv[1])
        return "\n".join(f"[{sid}] {self.titles[url]}: {url}" for url, sid in by_id) or "None."

The Agent Prompt

The prompt the planner sees every round. Three things are doing the work: a strict action contract of exactly one JSON object per turn, an enumerate-then-verify strategy, and hard evidence rules.

PLANNER_SYSTEM = "Choose the next search action. Return exactly one JSON object, no prose."

AGENT_PROMPT = """You are a deep search agent. You answer broad, fact-heavy questions by
iteratively searching the web and reasoning over the evidence you retrieve.

## Workflow
1. SEARCH  - issue up to {fanout} queries that attack the question from different angles.
2. ANALYZE - decide what is now supported and what is still missing.
3. REFINE  - search again only for genuinely missing facts, disambiguation, or verification.
4. FINISH  - stop as soon as the evidence can support a complete answer.

## Search strategy
- Enumerate candidates first, then verify each candidate's attributes.
- Prefer official, primary, or list/table sources.
- Never repeat a query string, and avoid synonym-only rewrites.

## Evidence rules
- Answer only from what retrieved sources explicitly state; no outside knowledge.
- If a source does not explicitly support a required fact, treat that fact as missing.

## Action contract - every turn returns exactly one JSON object
- {{"action": "search", "queries": ["...", "..."], "reason": "why these"}}
- {{"action": "finish", "reason": "why the evidence is sufficient"}}

## Question
{query}

## Evidence transcript
{evidence}"""

Model Calls

A plain OpenAI-compatible chat call, with two Mercury-specific notes:

reasoning_effort is the speed/quality dial: instant, low, medium, or high.Use response_format={"type": "json_object"} for the planner only. The final answer is free-form markdown, and forcing JSON there would coerce a perfectly good table into an empty object.

async def call(client, model, system, prompt, *, effort, max_tokens, as_json):
    kwargs = {
        "model": model,
        "messages": [{"role": "system", "content": system},
                     {"role": "user", "content": prompt}],
        "max_tokens": max_tokens,
        "temperature": 0.7,
    }
    if as_json:
        kwargs["response_format"] = {"type": "json_object"}
    if effort:
        kwargs["extra_body"] = {"reasoning_effort": effort}
    t0 = time.perf_counter()
    resp = await client.chat.completions.create(**kwargs)
    return (resp.choices[0].message.content or ""), int((time.perf_counter() - t0) * 1000)


def parse_action(text: str) -> dict:
    """Planner output -> action dict. Tolerates code fences and stray prose."""
    try:
        return json.loads(text)
    except json.JSONDecodeError:
        m = re.search(r"\{.*\}", text, re.S)
        return json.loads(m.group(0)) if m else {"action": "finish", "reason": "unparseable plan"}

def normalize_citations(text: str) -> str:
    """Rewrite full-width citation brackets to ASCII: 【1】 -> [1]."""
    return re.sub(r"[【\[]\s*(\d{1,3})\s*[】\]]", r"[\1]", text)


def clean_queries(raw: list, seen: list[set[str]], fanout: int) -> list[str]:
    out = []
    for item in raw:
        query = " ".join(str(item).split())
        words = set(re.findall(r"[a-z0-9]+", query.lower()))
        if len(query) < 12 or len(words) < 3:
            continue                                     # fragment, not a query
        if any(len(words & prev) / len(words | prev) > 0.8 for prev in seen):
            continue                                     # near-dupe of one we already ran
        seen.append(words)
        out.append(query)
        if len(out) >= fanout:
            break
    return out

Finally, a wall-clock ledger, so a run can show how much of its time was the model and how much was waiting on the web.

The Timings ledger

class Timings:
    def __init__(self):
        self.t0 = time.perf_counter()
        self.plan_ms = self.plan_calls = 0
        self.search_ms = self.search_rounds = 0
        self.synth_ms = 0

    def plan(self, ms: int) -> None:
        self.plan_ms += ms; self.plan_calls += 1

    def searched(self, ms: int) -> None:
        self.search_ms += ms; self.search_rounds += 1

    def synth(self, ms: int) -> None:
        self.synth_ms = ms

    def table(self, *, backend: str, model: str) -> str:
        total = max(1, int((time.perf_counter() - self.t0) * 1000))
        model_ms = self.plan_ms + self.synth_ms
        pct = lambda ms: f"{100 * ms / total:.0f}%"
        return (
            f"### {total / 1000:.1f} s total\n\n"
            f"| stage | calls | time | share |\n|---|---|---|---|\n"
            f"| {model} planning | {self.plan_calls} | {self.plan_ms:,} ms "
            f"| {pct(self.plan_ms)} |\n"
            f"| web search ({backend}) | {self.search_rounds} | {self.search_ms:,} ms "
            f"| {pct(self.search_ms)} |\n"
            f"| {model} synthesis | 1 | {self.synth_ms:,} ms | {pct(self.synth_ms)} |\n\n"
            f"**{model} took {model_ms / 1000:.1f} s of the {total / 1000:.1f} s "
            f"({pct(model_ms)}) — the other {pct(total - model_ms)} was spent "
            f"waiting on the web.**"
        )

The Loop

Plan, fan out searches in parallel, append evidence, and repeat until the agent says finish or the round cap is hit. Then one grounded synthesis call:

async def deep_search(
    query: str,
    *,
    columns: str | None = None,   # e.g. "State,Law name,Effective date" -> markdown table
    rounds: int = 3,              # max search rounds; the agent may stop earlier
    fanout: int = 3,              # parallel queries per round
    results: int = 5,             # search results per query
    backend: str | None = None,   # "exa" | "parallel"; defaults to SEARCH_BACKEND
    model: str = MODEL,
    effort: str = "instant",
    verbose: bool = True,
) -> str:
    searcher = AsyncOpenAI(api_key=os.environ["INCEPTION_API_KEY"], base_url=INCEPTION_BASE_URL)

    backend = backend or SEARCH_BACKEND
    timings = Timings()
    evidence = Evidence()
    seen_queries: list[set[str]] = []      # token sets of every query already issued
    finish_reason = f"reached the {rounds}-round cap"

    async with httpx.AsyncClient(timeout=60) as http:
        for round_no in range(1, rounds + 1):
            prompt = AGENT_PROMPT.format(
                fanout=fanout, query=query, evidence=evidence.transcript()
            ) + (
                f"\n\n## Next action\nRounds used: {round_no - 1}   "
                f"Rounds left: {rounds - round_no + 1}   Max queries this round: {fanout}\n"
                "Return exactly one JSON action now."
            )
            raw, ms = await call(searcher, model, PLANNER_SYSTEM, prompt,
                                 effort=effort, max_tokens=2048, as_json=True)
            timings.plan(ms)
            action = parse_action(raw)

            if action.get("action") != "search":
                finish_reason = str(action.get("reason") or "planner chose to finish")
                print(f"  round {round_no}: finish ({ms} ms) — {finish_reason}")
                break

            queries = clean_queries(action.get("queries") or [], seen_queries, fanout)
            if not queries:
                finish_reason = "planner produced no new usable queries"
                print(f"  round {round_no}: finish ({ms} ms) — {finish_reason}")
                break

            print(f"  round {round_no}: planned in {ms} ms — {len(queries)} queries")
            if verbose:
                for q in queries:
                    print(f"      · {q}")

            t0 = time.perf_counter()
            found = await search(http, queries, objective=query,
                                 num_results=results, backend=backend)
            search_ms = int((time.perf_counter() - t0) * 1000)
            timings.searched(search_ms)
            new = evidence.add_round(round_no, found)
            print(f"      searched in {search_ms} ms via {backend} "
                  f"— {new} new sources ({len(evidence.ids)} total)")

    shape = (
        f"Return a markdown table only — no prose before or after it.\n"
        f"Columns, exactly these and in this order: {' | '.join(columns.split(','))}\n"
        f"Emit one row per item the question asks for; leave a cell blank rather than guessing."
        if columns else
        "Answer in markdown, and cite the sources you used inline with plain square "
        "brackets, e.g. [1] or [2][5]."
    )
    final_prompt = (
        f"Question:\n{query}\n\n## Accumulated evidence\n\n{evidence.transcript()}\n\n"
        f"## Search phase conclusion\n\n{finish_reason}\n\n"
        f"## Sources\n\n{evidence.source_list()}\n\n## Instructions\n\n{shape}\n"
        "Use only the accumulated evidence above."
    )
    print(f"\n  synthesizing with {model} …")
    answer, ms = await call(searcher, model, "Answer only from the accumulated evidence.",
                            final_prompt, effort="", max_tokens=8192, as_json=False)
    timings.synth(ms)
    print(f"  answered in {ms} ms")
    return (f"{normalize_citations(answer.strip())}\n\n---\n"
            f"{timings.table(backend=backend, model=model)}\n\n---\n"
            f"**Sources**\n\n{evidence.source_list()}")

Run It

QUESTION = ("Which U.S. states have enacted a comprehensive consumer data privacy law? "
            "For each state, give the name of the law and the date it takes effect.")

answer = await deep_search(QUESTION, columns="State,Law name,Effective date", rounds=3)

display(Markdown(answer))

Three rounds of planning, 32 sources, and a synthesized table in about 7.5 seconds:

  round 1: planned in 302 ms — 3 queries
      · list of US states comprehensive consumer data privacy law effective date
      · state consumer privacy law comprehensive name effective date
      · comprehensive consumer data privacy statutes US states effective dates
      searched in 1279 ms via exa — 8 new sources (8 total)
  round 2: planned in 297 ms — 3 queries
      · list of U.S. states with comprehensive consumer data privacy laws effective dates 2026
      · official state websites consumer privacy law effective date
      · comprehensive consumer privacy law name and effective date per state
      searched in 1589 ms via exa — 10 new sources (18 total)
  round 3: planned in 324 ms — 3 queries
      · official state website California Consumer Privacy Act effective date
      · official state website Virginia Consumer Data Protection Act effective date
      · official state website Colorado Privacy Act effective date
      searched in 1679 ms via exa — 14 new sources (32 total)

  synthesizing with mercury-2 …
  answered in 1925 ms

[total 7.5s]

Planning cost roughly 900 ms across all three rounds. The rest was spent waiting on the web, which is the point: at Mercury speeds, the agent’s own deliberation stops being the bottleneck, so you can afford more rounds of it.

Your Own Question

answer = await deep_search(
    "What are the main obligations for general-purpose AI models under the EU AI Act, "
    "and when do they start applying?",
    rounds=2,
    verbose=False,
)
display(Markdown(answer))

Getting Started

API reference

Capabilities

Resources

Support

Search Agent: Mercury + Web Search

Run the Search Agent

Dependencies

API Keys

The Search Backend

Evidence Bookkeeping

The Agent Prompt

Model Calls

The Loop

Run It

Your Own Question

Run the Search Agent

​Dependencies

​API Keys

​The Search Backend

​Evidence Bookkeeping

​The Agent Prompt

​Model Calls

​The Loop

​Run It

​Your Own Question

Dependencies

API Keys

The Search Backend

Evidence Bookkeeping

The Agent Prompt

Model Calls

The Loop

Run It

Your Own Question