Build an AI Agent Web Search Tool with Source Tracking

An agent web search tool should return evidence records, not a concatenated string with its sources removed. Define a bounded search-and-read operation, preserve each URL and retrieval status, and keep failed reads out of the answer context. Tool calling decides when to invoke it; your application owns the execution contract.

For generated reports, choose the Markdown versus JSON consumer contract according to whether the consumer is a person or software.

When assembling a research workflow, use the research API evidence stack to keep discovery, extraction and source-to-claim review separate.

Define what the consumer can trust

Do not convert a search snippet into a source body. Choose which URLs the application may read, then validate the Reader result separately. An exact-host allowlist is useful for a controlled documentation task; it is not a universal policy for open-ended web research. In a deployed service, also apply your outbound URL and network-access policy.

Implement the tool as an ordinary tested function

The example uses SearchCans search and Reader with POST requests. It reads data.organic and link, deduplicates candidates and keeps at most two sources. Install requests and set SEARCHCANS_API_KEY. Inject a fake session for deterministic testing rather than calling a model to test parsing.

import os
import requests
from datetime import datetime, timezone
from urllib.parse import urlsplit, urldefrag

def organic(envelope):
    if envelope.get("code") != 0:
        raise ValueError("Search business error")
    data = envelope.get("data")
    if not isinstance(data, dict) or not isinstance(data.get("organic"), list):
        raise ValueError("Expected data.organic array")
    return [row for row in data["organic"] if isinstance(row, dict)
            and isinstance(row.get("link"), str)
            and urlsplit(row["link"]).scheme == "https"
            and urlsplit(row["link"]).hostname
            and not urlsplit(row["link"]).username]

def call(path, payload, session=requests, timeout=40):
    response = session.post("https://www.searchcans.com/api/v1/" + path,
        headers={"Authorization": "Bearer " + os.environ["SEARCHCANS_API_KEY"]},
        json=payload, timeout=timeout)
    response.raise_for_status()
    envelope = response.json()
    if envelope.get("code") != 0 or not isinstance(envelope.get("data"), dict):
        raise ValueError("Business or response-schema error")
    return envelope

import time

def research_tool(query, allowed_hosts, budget_seconds=50, session=requests):
    if budget_seconds <= 0:
        raise ValueError("Positive local time budget required")
    deadline = time.monotonic() + budget_seconds
    remaining = lambda: max(0, deadline - time.monotonic())
    envelope = call("search", {"t":"google", "s":query,
        "country":"us", "language":"en", "d":10000}, session,
        timeout=min(remaining(), 15))
    urls = list(dict.fromkeys(row["link"] for row in organic(envelope)
        if urlsplit(row["link"]).hostname in allowed_hosts))[:2]
    records = []
    for url in urls:
        record = {"source_url":url, "attempted_at":datetime.now(timezone.utc).isoformat()}
        if remaining() < 15:
            record.update(status="budget_exhausted", text="")
        else:
            try:
                data = call("url", {"t":"url", "s":url, "mode":0,
                    "proxy":0, "maxCache":0, "d":10000}, session, timeout=15)["data"]
                markdown = data.get("markdown")
                if not isinstance(markdown, str) or not markdown.strip():
                    raise ValueError("Empty source")
                record.update(status="ok", text=markdown, retrieved_at=datetime.now(timezone.utc).isoformat())
            except (requests.RequestException, ValueError):
                record.update(status="read_failed", text="")
        records.append(record)
    return {"query":query, "status":"ok" if records and all(r["status"]=="ok" for r in records)
            else "partial" if any(r["status"]=="ok" for r in records) else "no_usable_sources",
            "sources":records}

Understand the time budget

budget_seconds is a local admission budget: the function stops starting more reads when too little time remains. A requests timeout governs connection/read waiting, not a strict end-to-end deadline, so this example must not be sold as a hard wall-clock guarantee. Use an async transport with an outer task deadline if your service needs enforceable cancellation. Retry individual transient failures within that deadline; never silently rerun an entire research task.

Test partial results and missing evidence

A synthetic search returns two approved URLs. Make the first Reader call return a nonempty body and the second return a business error. The result must be partial, contain both URLs and keep the failed source text empty. Also test an empty organic array, a row without link, a forbidden host and malformed Reader data. The fixtures test the tool contract; they do not establish answer correctness.

Register the function with your chosen agent framework using a narrow description: search approved documentation and return source records. Keep credentials outside the model-visible arguments. Treat retrieved page text as untrusted content and require source-to-claim checks before using it in an answer.

Use the grounding architecture for the broader agent loop, the evidence-support checks for claim quality, and the API failure contract for errors. This page implements the tool boundary rather than another supplier-selection guide.