Build Production-Ready AI Agents with the You.com Web Search API

TLDR: To make an AI agent production-ready on the You.com Web Search API, send every call through one gateway that limits request rate, honors Retry-After, caches on the full request, caps spend, and logs search_uuid. Request capped highlights instead of raw snippets, never retry 402, 403, or 422 responses, and test the failure paths offline. The tested gateway below does all of it in standard-library Python.

A prototype agent calls a search tool and prints an answer. A production agent also has to survive concurrent users, a monthly invoice, a security review, and the support ticket that asks where an answer came from. Quickstarts optimize for time to first successful request, so none of that appears in them.

This guide covers that gap for agents built on the You.com Web Search API. It assumes you can already make a call and stays on operations: failure handling, rate limits, prompt size, spend, and traceability. For the tool definition, see How to Add Web Search to Agent Tool Calling. For loop design, see What Are AI Agent Architecture Patterns? For scoring answers, see How to Run an AI Agent Evaluation.

What breaks first when a search agent meets real traffic?

Agents are bursty, and the burst is what breaks them. Take an agent that fans out 4 subqueries per turn, runs 5 turns per task, and finishes 500 tasks a day. That is 10,000 calls a day, an average of about 0.12 requests per second. But the 4 subqueries leave together, so 15 tasks in flight send 60 calls at once. The self-serve default on /v1/search is 10 requests per second, so the peak decides whether users see errors, not the daily average.

FailureWhat the docs sayWhat the gateway doesRate limit under fan-out10 requests per second by default on self-serve accounts. Responses carry X-RateLimit-* headers.Limits its own rate and pauses every worker on a 429Prompt bloatSnippets run 100 to 200 words per result. Uncapped highlights add up to roughly 6,000 to 7,500 tokens per 10 results.Requests highlights with a token capOut of creditsA 402 means Insufficient credits. Auto Top-Up avoids stalls but has no monthly cap yet.Stops retrying and enforces its own budgetCost growthOne request is one billable call. Full-page extraction adds $1.00 per 1,000 pages crawled live.Prices the worst case before sending, and cachesUntraceable answersEvery response includes metadata.search_uuid.Logs it and returns it with each passage

What does a production architecture look like?

Hold one rule: the agent never calls the API directly. Every call passes through a gateway you own, so the limiter, cache, budget, and logs live in one file instead of in every tool wrapper.

user query
 -> agent core: plan, act, observe
 -> search gateway: limiter, cache, budget, retries, logs
 -> You.com Web Search API: POST https://ydc-index.io/v1/search
 -> grounding stage: highlights, dedupe, source URLs, search_uuid
 -> model answer with citations


The Web Search API does retrieval only, so it returns ahead of the Answer API. The pricing page lists a p50 of about 0.4 seconds for Web Search and about 2.7 seconds for Answer (as of October 2026).

Which host and key does each API need?

You.com runs five APIs. Web Search and Contents live on ydc-index.io, while Answer, Research, and Finance Research run only on api.you.com. All five use the X-API-Key header, but keys are scoped per product. A Web Search-only key gets 403 Missing required scopes from /v1/answer, and sending Answer, Research, or Finance Research traffic to ydc-index.io returns 403 Missing Authentication Token even with a valid key. The API chooser compares latency and price for all five.

How should the gateway handle rate limits, retries, and errors?

Each status code calls for a different response, and only some failures can succeed on a retry. The error reference lists the bodies.

StatusWhat it meansGateway action429Over the per-second limitWait for Retry-After, else back off exponentially, then retry500 or no responseServer or network failureRetry a few times with backoff, then raise Unavailable402Insufficient credits on a keyed accountRaise OutOfCredits and alert a human. Do not retry401, 403Bad key, missing scope, or wrong hostRaise Misconfigured. Do not retry400, 422Invalid parameter or combinationRaise BadRequest. Do not retry

RFC 9110 defines Retry-After as a number of seconds or an HTTP date. The You.com docs say to honor it when present without naming the form, so the gateway parses both and falls back to exponential backoff, capped at 60 seconds. The docs prescribe backoff only for 429, so retrying 5xx responses and network failures is this gateway's own policy.

Every response also carries X-RateLimit-Limit, X-RateLimit-Remaining, and X-RateLimit-Reset, a Unix timestamp for the end of the window. The rate limits page says to throttle on them rather than wait for a 429, so the gateway pauses until the reset when Remaining hits zero, capped at 60 seconds in case a bad value arrives.

A token bucket smooths bursts. It sends at most burst + rate requests in any one-second window, so the defaults of 6 and 2 stay at 8, under the self-serve limit of 10, with room for retries. Treat the limit as shared by every worker: a 429 on one thread sets a pause that all threads honor. Across processes you need a shared limiter. Enterprise plans run on custom limits set in the contract, so ask before launch if your peak needs more.

Keep the client timeout above the crawl time you allow. Full-page extraction waits up to crawl_timeout per page (10 seconds by default, 60 at most), so raise the gateway's 15-second timeout along with it.

Save the next two blocks in order as search_gateway.py, and add the grounding function from the next section at the end. It needs Python 3.9 or later and only the standard library. The first block holds the errors, limiter, cache, and cost estimate.

import hashlib
import json
import random
import sys
import threading
import time
import urllib.error
import urllib.request
from collections import OrderedDict
from email.utils import parsedate_to_datetime

SEARCH_URL = "https://ydc-index.io/v1/search"
PRICE_PER_CALL = 0.005        # $5.00 per 1,000 Web Search calls
PRICE_PER_LIVE_PAGE = 0.001   # $1.00 per 1,000 pages crawled live (full_page only)


class GatewayError(Exception): pass
class BudgetExceeded(GatewayError): pass   # local cap reached, nothing was sent
class OutOfCredits(GatewayError): pass     # 402: retrying cannot help
class Misconfigured(GatewayError): pass    # 401 or 403: key, product scope, or host is wrong
class BadRequest(GatewayError): pass       # 400, 422, other 4xx: a bug in the calling code
class Unavailable(GatewayError): pass      # retries exhausted on 429, 5xx, or no response


class TokenBucket:
   """In any one-second window this sends at most burst + rate requests,
   so keep that sum under your plan's limit (self-serve default: 10 per second)."""

   def __init__(self, rate, burst=2, sleep=time.sleep, clock=time.monotonic):
       self.rate, self.burst, self.sleep, self.clock = rate, burst, sleep, clock
       self.tokens, self.stamp, self.lock = float(burst), clock(), threading.Lock()

   def acquire(self):
       while True:
           with self.lock:
               now = self.clock()
               self.tokens = min(self.burst, self.tokens + (now - self.stamp) * self.rate)
               self.stamp = now
               if self.tokens >= 1 - 1e-9:  # tolerate float rounding
                   self.tokens = max(self.tokens - 1, 0.0)
                   return
               wait = (1 - self.tokens) / self.rate
           self.sleep(wait)


class TTLCache:
   def __init__(self, ttl, max_items=1000, clock=time.monotonic):
       self.ttl, self.max_items, self.clock = ttl, max_items, clock
       self.items, self.lock = OrderedDict(), threading.Lock()

   def get(self, key):
       with self.lock:
           entry = self.items.get(key)
           if entry and entry[0] > self.clock():
               self.items.move_to_end(key)
               return entry[1]
           self.items.pop(key, None)
           return None

   def put(self, key, value):
       with self.lock:
           self.items[key] = (self.clock() + self.ttl, value)
           self.items.move_to_end(key)
           while len(self.items) > self.max_items:
               self.items.popitem(last=False)


def worst_case_cost(body):
   """Upper bound in USD for one request, from the published prices."""
   cost = PRICE_PER_CALL
   extraction = body.get("extraction") or {}
   if extraction.get("extraction_mode") == "full_page" and extraction.get("extraction_source") != "cache":
       # full_page crawls web AND news results: up to `count` pages in each section
       cost += PRICE_PER_LIVE_PAGE * 2 * body.get("count", 10)
   return cost


The gateway runs a fixed order: check the cache, price the worst case, take a limiter token, send, classify the status, and retry only what can succeed.

class SearchGateway:
   def __init__(self, api_key, url=None, rate=6, burst=2, cache_ttl=300, budget_usd=50.0,
                max_attempts=5, timeout=15, log=None, sleep=time.sleep, wall=time.time):
       if not api_key:
           raise ValueError("api_key is required")
       self.api_key, self.url, self.timeout = api_key, url or SEARCH_URL, timeout
       self.bucket = TokenBucket(rate, burst, sleep=sleep)
       self.cache = TTLCache(cache_ttl)
       self.budget, self.spent, self.max_attempts = budget_usd, 0.0, max_attempts
       self.log = log or (lambda record: print(json.dumps(record), file=sys.stderr))
       self.sleep, self.wall = sleep, wall
       self.pause_until, self.lock = 0.0, threading.Lock()

   def search(self, query, **params):
       body = {"query": query, **params}
       key = json.dumps(body, sort_keys=True)  # the query plus every filter and extraction setting
       started, cost = time.monotonic(), worst_case_cost(body)
       cached = self.cache.get(key)
       if cached is not None:  # raw bytes, so callers cannot mutate what the cache holds
           data = json.loads(cached)
           self._log(body, data, 0, "cache_hit", 0.0)
           return data
       for attempt in range(1, self.max_attempts + 1):
           self._wait_for_pause()
           with self.lock:
               if self.spent + cost > self.budget:
                   raise BudgetExceeded("cap $%.2f, spent $%.3f" % (self.budget, self.spent))
               self.spent += cost  # counts every attempt, so it over-counts rather than under-counts
           self.bucket.acquire()
           status, headers, payload = self._post(body)
           self._note_rate_limit(headers)
           if status == 200:
               self.cache.put(key, payload)
               data = json.loads(payload)
               self._log(body, data, attempt, "ok", time.monotonic() - started)
               return data
           if status == 402:
               raise OutOfCredits(payload[:200])
           if status in (401, 403):
               raise Misconfigured("%s %s" % (status, payload[:200]))
           if status != 0 and status != 429 and status < 500:
               raise BadRequest("%s %s" % (status, payload[:200]))
           if attempt < self.max_attempts:  # 429, 5xx, or no response at all
               delay = self._delay(headers, attempt)
               if status == 429:
                   self.pause_until = max(self.pause_until, self.wall() + delay)  # pauses every thread
               else:
                   self.sleep(delay)
       raise Unavailable("gave up after %d attempts (last status %s)" % (self.max_attempts, status))

   def _post(self, body):
       request = urllib.request.Request(
           self.url, data=json.dumps(body).encode(), method="POST",
           headers={"X-API-Key": self.api_key, "Content-Type": "application/json"})
       try:
           with urllib.request.urlopen(request, timeout=self.timeout) as response:
               return response.status, response.headers, response.read()
       except urllib.error.HTTPError as err:
           return err.code, err.headers, err.read()
       except OSError:  # connection refused, DNS failure, timeout
           return 0, {}, b""

   def _delay(self, headers, attempt):
       value = (headers.get("Retry-After") or "").strip()
       try:  # delay-seconds
           seconds = float(value)
           if 0 <= seconds < float("inf"):
               return min(seconds, 120.0)
       except ValueError:
           pass
       try:  # HTTP-date
           return min(max(parsedate_to_datetime(value).timestamp() - self.wall(), 0.0), 120.0)
       except (TypeError, ValueError):
           return min(2 ** (attempt - 1), 60) + random.uniform(0, 0.5)  # capped exponential backoff

   def _note_rate_limit(self, headers):
       try:  # X-RateLimit-Reset is a Unix timestamp
           if float(headers.get("X-RateLimit-Remaining", "1")) <= 0:
               reset = float(headers["X-RateLimit-Reset"])
               self.pause_until = max(self.pause_until, min(reset, self.wall() + 60))
       except (KeyError, ValueError):
           pass

   def _wait_for_pause(self):
       wait = self.pause_until - self.wall()
       if wait > 0:
           self.sleep(wait)

   def _log(self, body, data, attempts, event, elapsed):
       meta = data.get("metadata") or {}
       self.log({"event": event, "query_hash": hashlib.sha256(body["query"].encode()).hexdigest()[:12],
                 "search_uuid": meta.get("search_uuid"), "api_latency_s": meta.get("latency"),
                 "elapsed_s": round(elapsed, 3), "attempts": attempts, "spent_usd": round(self.spent, 4)})


Two design choices are worth copying even if you rewrite the rest:

• The cache stores raw response bytes, so a caller that edits its copy cannot corrupt the next hit.
• The budget counts every attempt, retries included, at the worst-case price, so it over-counts rather than under-counts.

These blocks were tested offline: 25 unit tests on Python 3.9 against a local mock server and a fake clock, with no real key and no live calls. They cover every status class in the table, both Retry-After forms, backoff and its ceiling, budget refusal before any request, header pauses, cache keys and expiry, a pause shared across threads, and a limiter that never exceeds 8 sends in a second. They prove the client logic, not live rate-limit behavior, billing, or relevance, so run your own authorized integration test before launch.

How do you keep prompts small and answers citable?

Each web result can carry three levels of text, and the default suits a results page better than an agent.

LevelHow to request itFieldWhat you getSnippetsDefault, no extractionsnippets100 to 200 words of short, keyword-centered fragments per result, suited to a person skimming resultsHighlightsextraction.extraction_mode: "highlights"contents.highlightsThe passages that address the query, and nothing elseFull pageextraction.extraction_mode: "full_page"contents.markdown or contents.htmlThe whole page, at $1.00 per 1,000 pages crawled live

For grounding an agent, use highlights and cap them. Uncapped, highlights for a 10-result search add up to roughly 6,000 to 7,500 tokens. extraction.highlights.total_max_tokens accepts 1,024 to 8,192 and shares the budget across the web results, so count decides the split: at 1,024, a count of 10 typically returns five to 10 short results and a count of 3 returns three longer ones. The cap is approximate, so count tokens yourself if the prompt has a hard limit. Values outside the range return a 422, and highlights mode rejects crawl_timeout. The page content guide has the full parameter table.

curl -X POST https://ydc-index.io/v1/search \
 -H "X-API-Key: $YDC_API_KEY" \
 -H "Content-Type: application/json" \
 -d '{
   "query": "federal reserve interest rate decision",
   "count": 5,
   "extraction": {
     "extraction_mode": "highlights",
     "highlights": {"total_max_tokens": 2048}
   }
 }'


Put domain filters and freshness in the request, not the prompt. include_domains and boost_domains take up to 500 domains each, and exclude_domains up to 5,000. include_domains cannot be combined with the other two (the API returns a 422), but boost_domains pairs with exclude_domains. freshness accepts day, week, month, year, or a YYYY-MM-DDtoYYYY-MM-DD range. Use POST for long lists, because GET runs into URL length limits. The request controls guide covers the rest.

Flatten the response in one place. Add this function to the end of the same file. to_evidence prefers highlights, falls back to snippets, drops duplicate URLs and empty results, and carries page_age and search_uuid with each passage. It never iterates results directly, because results.web and results.news are both optional. News results carry a description rather than highlights, so it reads web results only. On web results, page_age is defined only as the age of the search result, so show it but do not treat it as a crawl time.

def to_evidence(response, per_result=2):
   """Flatten a response into numbered, citable passages. Never iterate `results` directly."""
   meta = response.get("metadata") or {}
   web = (response.get("results") or {}).get("web") or []
   evidence, seen = [], set()
   for hit in web:
       url = hit.get("url")
       passages = (hit.get("contents") or {}).get("highlights") or hit.get("snippets") or []
       if not url or url in seen or not passages:
           continue
       seen.add(url)
       evidence.append({"n": len(evidence) + 1, "url": url, "title": hit.get("title"),
                        "page_age": hit.get("page_age"), "passages": passages[:per_result],
                        "search_uuid": meta.get("search_uuid")})
   return evidence


A finance-news agent's search tool then reduces to a few lines. It degrades when the budget is spent or retries run out, and escalates when the account is out of credits:

import logging
import os

from search_gateway import BudgetExceeded, OutOfCredits, SearchGateway, Unavailable, to_evidence

gateway = SearchGateway(os.environ["YDC_API_KEY"], budget_usd=25.0)


def search_for_agent(query):
   try:
       response = gateway.search(
           query,
           count=5,
           freshness="week",
           include_domains=["reuters.com", "apnews.com", "ft.com", "bloomberg.com"],
           extraction={"extraction_mode": "highlights", "highlights": {"total_max_tokens": 2048}},
       )
   except (BudgetExceeded, Unavailable):
       return []  # degrade: answer without fresh search, or queue the task
   except OutOfCredits:
       logging.critical("You.com credits exhausted; page a human")
       raise
   return to_evidence(response)


Pass an empty list to the model as an explicit "no sources found" result, so it says so instead of guessing. Treat every passage as untrusted input: a web page can carry instructions aimed at your model, so keep passages in a delimited evidence block and never let them change system rules. When the Web Page Fights Back covers the attack.

Store the search_uuid beside every answer. After a cache hit it is the uuid of the original request, which is what an audit needs, because it names the retrieval the answer actually used. For how citations are attached and checked, see What Are LLM Citations?

How do you control what the agent spends?

Web Search is billed per request: $5.00 per 1,000 calls, whether a call returns 1 result or 100 (as of October 2026). Full-page extraction adds $1.00 per 1,000 pages crawled live, and it crawls every web and news result, so a count of 10 can fetch 20 pages: a worst case of $0.025 per call against $0.005 for the search alone. The default blend source charges only for pages it cannot serve from cache, and extraction_source: "cache" adds nothing. The billing page has the full tables.

Using the 10,000 calls a day from earlier, with 40% assumed to hit your cache (use your own rate):

SetupBillable calls per dayCost per daySnippets or highlights, no cache10,000$50.00Snippets or highlights, 40% cache hits6,000$30.00Full page at the worst case, no cache10,000up to $250.00Full page at the worst case, 40% cache hits6,000up to $150.00

Key the cache on the whole request. A cache keyed on the query alone serves results fetched with a different count, freshness window, filter, or extraction mode, a correctness bug that looks like a relevance bug. The docs suggest a 5 to 15 minute TTL for most use cases, and the gateway defaults to 5. Shorten it for news-style queries and lengthen it for stable lookups. Real-Time Web Search API covers caching and freshness.

Auto Top-Up, which the billing page recommends for production workloads, buys credits when the balance falls below a threshold you set. The page lists monthly caps, frequency limits, and per-charge maximums as planned but not yet exposed. So with Auto Top-Up on, nothing on the account stops a runaway loop from triggering purchase after purchase, and with it off the balance is the ceiling and the next call returns a 402. Either way, the cap that stops a bad loop has to live in your code. The gateway prices each request at its worst case before sending and raises BudgetExceeded instead of calling. It cannot see what the account actually paid, so reconcile it against the Account Balance endpoint, which returns cents:

curl -sS -X GET "https://api.you.com/v1/billing/account_balance" \
 -H "X-API-Key: $YDC_API_KEY" \
 -H "Accept: application/json"


Divide data.attributes.balance by 100 for dollars, poll it from a scheduled job, and alert well before zero, whether or not Auto Top-Up is on.

What changes when the agent reaches You.com through MCP?

Connecting an MCP client skips the request layer, so it also skips your gateway, and the controls move into server configuration. The hosted server at https://api.you.com/mcp exposes four tools by default: you-search, you-contents, you-balance, and you-discover. you-research and you-finance stay off until you use the /mcp/research or /mcp/finance path or list them in ?tools=. The keyless ?profile=free exposes only you-search and you-discover, at 100 queries per day. There is no Answer tool, so an agent that needs a one-pass cited answer calls the Answer API over REST. The MCP server guide lists every option.

The docs suggest separate server entries with separate API keys for the expensive tools, so a research call can be budgeted and revoked on its own:

{
 "mcpServers": {
   "you.com_search": {
     "url": "https://api.you.com/mcp?tools=you-search,you-contents",
     "headers": {"Authorization": "Bearer <SEARCH_KEY>"}
   },
   "you.com_research": {
     "url": "https://api.you.com/mcp/research",
     "headers": {"Authorization": "Bearer <RESEARCH_KEY>"}
   }
 }
}


Authentication differs from REST. MCP takes Authorization: Bearer or OAuth 2.1, while the REST APIs take X-API-Key. For when to use MCP and when to use native tool calling, see How to Add Web Search to Agent Tool Calling.

What should you test and log before launch?

• Failure paths. Test 429 with and without Retry-After, 402, 403, 422, and a dropped connection. Build a mock server and a fake clock first, because they make retry timing testable without waiting.
• Per-call logs. Record search_uuid, the API-reported latency, your own end-to-end time, attempts, cache hit or miss, and a hash of the query when queries are sensitive.
• Latency. Alert on the p95 of the whole trajectory, not the mean, because parallel subqueries make a turn wait for the slowest one. What Is API Latency? covers measurement.
• Quality. Gate changes to extraction mode, count, and filters on an eval set built from your own logged queries.
• Access. Smoke-test every API the agent can reach, with the key it will use, in each environment, so a missing scope or wrong host fails in staging.
• Data handling. If your security review requires zero data retention, it is available on enterprise agreements for the Web Search and Answer APIs, not yet for Contents or Research. See Zero Data Retention.

How do you get started?

Create an account and an API key at you.com/platform and export it as YDC_API_KEY, the variable name the docs, SDKs, and integrations all use. In production, keep it in a secrets manager and never in source. New accounts start with $100 in free credits, which the pricing page puts at 20,000 searches. With no account, the free MCP profile gives you 100 queries a day. Run a first request from the Web Search API quickstart, then put the gateway in front of it before anything else calls the API. Prefer the Python SDK? Web Search API in Python covers its timeouts and opt-in retries.