Google Search MCP Server

Instructions

Parameters:
    q: Query string. Trimmed; required.
    num: Number of results to return (1..10; clamped).
    start: 1-based index for pagination start (clamped to >=1).
    siteSearch: Limit results to a site (or domain) per CSE rules.
    siteSearchFilter: "i" to include or "e" to exclude `siteSearch`.
    safe: SafeSearch level: "off" or "active".
    gl: Geolocation/country code.
    hl: UI language.
    lr: Language restrict (e.g., "lang_en").
    useSiteRestrict: Use the siterestrict endpoint variant.
    dateRestrict: Time filter (e.g., "d7", "m3", "y1").
    exactTerms, orTerms, excludeTerms: Query modifiers.
    cxOverride: Override the configured CSE ID (avoid echoing to clients).
    lean_fields: If True, request a smaller response via fields projection.

Returns:
    A dict with keys: provider, query (sanitized), searchInfo, nextPage,
    latency_ms, results (normalized), raw (subset), trace (q hash).

Raises:
    ValueError: For invalid parameter values (e.g., unsupported safe).
    RuntimeError: For Google API errors or network failures.

Implementation Reference

• server.py:191-338 (handler)

  The primary handler for the 'search' tool. Decorated with @mcp.tool(), it defines the input parameters with types, docstring describing schema, performs API call to Google CSE, normalizes results, and returns structured data.

  @mcp.tool()
  async def search(
      q: str,
      num: int = 5,
      start: int = 1,
      siteSearch: str | None = None,
      siteSearchFilter: str | None = None,  # "i" include | "e" exclude
      safe: str | None = None,  # "off" | "active" (CSE)
      gl: str | None = None,
      hl: str | None = None,
      lr: str | None = None,
      useSiteRestrict: bool = False,
      dateRestrict: str | None = None,  # e.g. "d7", "m3", "y1"
      exactTerms: str | None = None,
      orTerms: str | None = None,
      excludeTerms: str | None = None,
      cxOverride: str | None = None,
      lean_fields: bool = True,  # shrink Google payload
  ) -> dict[str, Any]:
      """Google Programmable Search (CSE) via MCP.
  
      Parameters:
          q: Query string. Trimmed; required.
          num: Number of results to return (1..10; clamped).
          start: 1-based index for pagination start (clamped to >=1).
          siteSearch: Limit results to a site (or domain) per CSE rules.
          siteSearchFilter: "i" to include or "e" to exclude `siteSearch`.
          safe: SafeSearch level: "off" or "active".
          gl: Geolocation/country code.
          hl: UI language.
          lr: Language restrict (e.g., "lang_en").
          useSiteRestrict: Use the siterestrict endpoint variant.
          dateRestrict: Time filter (e.g., "d7", "m3", "y1").
          exactTerms, orTerms, excludeTerms: Query modifiers.
          cxOverride: Override the configured CSE ID (avoid echoing to clients).
          lean_fields: If True, request a smaller response via fields projection.
  
      Returns:
          A dict with keys: provider, query (sanitized), searchInfo, nextPage,
          latency_ms, results (normalized), raw (subset), trace (q hash).
  
      Raises:
          ValueError: For invalid parameter values (e.g., unsupported safe).
          RuntimeError: For Google API errors or network failures.
      """
      # --- Input hygiene
      q = q.strip()
      num = max(1, min(10, int(num)))
      start = max(1, int(start))
      if safe and safe not in {"off", "active"}:
          raise ValueError('safe must be "off" or "active" for Google CSE')
  
      # Endpoint selection
      endpoint = (
          GOOGLE_CSE_SITERESTRICT_ENDPOINT if useSiteRestrict else GOOGLE_CSE_ENDPOINT
      )
  
      # Base params
      params: dict[str, Any] = {
          "key": GOOGLE_API_KEY,
          "cx": cxOverride or GOOGLE_CX,
          "q": q,
          "num": num,
          "start": start,
      }
      # Optional params
      if siteSearch:
          params["siteSearch"] = siteSearch
      if siteSearchFilter:
          params["siteSearchFilter"] = siteSearchFilter  # "i" or "e"
      if safe:
          params["safe"] = safe
      if gl:
          params["gl"] = gl
      if hl:
          params["hl"] = hl
      if lr:
          params["lr"] = lr
      if dateRestrict:
          params["dateRestrict"] = dateRestrict
      if exactTerms:
          params["exactTerms"] = exactTerms
      if orTerms:
          params["orTerms"] = orTerms
      if excludeTerms:
          params["excludeTerms"] = excludeTerms
  
      # Lean response projection to save bandwidth (smaller payloads)
      if lean_fields:
          params["fields"] = (
              "items(title,link,snippet),"
              "queries(nextPage(startIndex)),"
              "searchInformation(searchTime,totalResults),"
              "kind"
          )
  
      t0 = time.perf_counter()
      data = await _cse_get(endpoint, params)
      dt = round((time.perf_counter() - t0) * 1000)
  
      # Optional query/latency logging (no secrets)
      if LOG_QUERIES:
          endpoint_name = (
              "siterestrict" if endpoint == GOOGLE_CSE_SITERESTRICT_ENDPOINT else "cse"
          )
          parts = [
              f"q_hash={_hash_q(q)}",
              f"dt_ms={dt}",
              f"num={num}",
              f"start={start}",
              f"safe={safe or '-'}",
              f"endpoint={endpoint_name}",
          ]
          if LOG_QUERY_TEXT:
              parts.append(f'q="{q}"')
          _logger.info("search %s", " ".join(parts))
  
      items = data.get("items") or []
      next_page = (data.get("queries", {}).get("nextPage") or [{}])[0].get("startIndex")
  
      # Avoid leaking secrets (e.g., API key/cx) in the echoed query payload
      echoed_query = {
          "q": q,
          "num": num,
          "start": start,
          "safe": safe,
          "gl": gl,
          "hl": hl,
          "lr": lr,
          "siteSearch": siteSearch,
          "siteSearchFilter": siteSearchFilter,
          "dateRestrict": dateRestrict,
          "exactTerms": exactTerms,
          "orTerms": orTerms,
          "excludeTerms": excludeTerms,
          # "cx": (cxOverride or GOOGLE_CX),  # omit unless you want it visible
      }
  
      return {
          "provider": "google-cse",
          "query": echoed_query,
          "searchInfo": data.get("searchInformation", {}),
          "nextPage": next_page,
          "latency_ms": dt,
          "results": _normalize(items),
          "raw": {"kind": data.get("kind")},
          "trace": {"q_hash": _hash_q(q)},
      }

• server.py:39-39 (registration)

  Creates the FastMCP server instance named 'google-search' to which the 'search' tool is registered via decorator.

  mcp = FastMCP(name="google-search")

• server.py:100-124 (helper)

  Helper function to normalize Google search results, applying domain filtering and formatting into {title, url, snippet, rank}.

  def _normalize(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
      """Transform raw Google CSE `items` into a concise result list.
  
      Applies optional domain allowlist filtering (ALLOW_DOMAINS) and returns
      dictionaries with shape: {title, url, snippet, rank}.
      """
      results: list[dict[str, Any]] = []
      for i, it in enumerate(items or [], start=1):
          url = it.get("link")
          if ALLOW_DOMAINS:
              # Skip results outside the allowlist
              from urllib.parse import urlparse
  
              host = (urlparse(url).netloc or "").lower()
              if not any(host.endswith(dom) for dom in ALLOW_DOMAINS):
                  continue
          results.append(
              {
                  "title": it.get("title"),
                  "url": url,
                  "snippet": it.get("snippet"),
                  "rank": i,
              }
          )
      return results

• server.py:132-189 (helper)

  Helper to perform HTTP GET to Google CSE API with retry logic for transient errors.

  async def _cse_get(endpoint: str, params: dict[str, Any]) -> dict[str, Any]:
      """Issue a GET to the Google CSE endpoint with small retries.
  
      Retries up to 3 times on common transient errors (429/5xx), with capped
      exponential backoff and jitter. Raises RuntimeError with status/text for
      non-retryable HTTP errors and wraps httpx exceptions on network errors.
      """
      # Retry small: 3 attempts on 429/5xx with jitter
      for attempt in range(3):
          try:
              r = await _http.get(endpoint, params=params)
              # Log status at DEBUG for observability (avoid logging params with secrets)
              if _logger.isEnabledFor(logging.DEBUG):
                  _logger.debug(
                      "cse_get attempt=%d status=%s", attempt + 1, r.status_code
                  )
              r.raise_for_status()
              return r.json()
          except httpx.HTTPStatusError as e:
              status = e.response.status_code
              retryable = status in (429, 500, 502, 503, 504) and attempt < 2
              # Respect Retry-After if present, else use capped backoff with jitter
              delay = None
              if retryable:
                  ra = e.response.headers.get("Retry-After")
                  if ra:
                      try:
                          delay = max(0.0, float(ra))
                      except ValueError:
                          delay = None
                  if delay is None:
                      delay = (0.2 + random.random() * 0.4) * (2**attempt)
                  if _logger.isEnabledFor(logging.DEBUG):
                      _logger.debug(
                          "cse_get retrying attempt=%d http_status=%s delay=%.2fs",
                          attempt + 1,
                          status,
                          delay,
                      )
                  await asyncio.sleep(delay)
                  continue
              # Bubble up a clean MCP error string
              detail = e.response.text[:500]
              raise RuntimeError(f"CSE request failed ({status}): {detail}") from e
          except httpx.HTTPError as e:
              if attempt < 2:
                  delay = (0.2 + random.random() * 0.4) * (2**attempt)
                  if _logger.isEnabledFor(logging.DEBUG):
                      _logger.debug(
                          "cse_get network error attempt=%d delay=%.2fs error=%s",
                          attempt + 1,
                          delay,
                          str(e),
                      )
                  await asyncio.sleep(delay)
                  continue
              raise RuntimeError(f"Network error contacting CSE: {e!s}") from e

• server.py:192-235 (schema)

  Input/output schema inferred from type annotations and detailed docstring describing parameters, return value, and exceptions.

  async def search(
      q: str,
      num: int = 5,
      start: int = 1,
      siteSearch: str | None = None,
      siteSearchFilter: str | None = None,  # "i" include | "e" exclude
      safe: str | None = None,  # "off" | "active" (CSE)
      gl: str | None = None,
      hl: str | None = None,
      lr: str | None = None,
      useSiteRestrict: bool = False,
      dateRestrict: str | None = None,  # e.g. "d7", "m3", "y1"
      exactTerms: str | None = None,
      orTerms: str | None = None,
      excludeTerms: str | None = None,
      cxOverride: str | None = None,
      lean_fields: bool = True,  # shrink Google payload
  ) -> dict[str, Any]:
      """Google Programmable Search (CSE) via MCP.
  
      Parameters:
          q: Query string. Trimmed; required.
          num: Number of results to return (1..10; clamped).
          start: 1-based index for pagination start (clamped to >=1).
          siteSearch: Limit results to a site (or domain) per CSE rules.
          siteSearchFilter: "i" to include or "e" to exclude `siteSearch`.
          safe: SafeSearch level: "off" or "active".
          gl: Geolocation/country code.
          hl: UI language.
          lr: Language restrict (e.g., "lang_en").
          useSiteRestrict: Use the siterestrict endpoint variant.
          dateRestrict: Time filter (e.g., "d7", "m3", "y1").
          exactTerms, orTerms, excludeTerms: Query modifiers.
          cxOverride: Override the configured CSE ID (avoid echoing to clients).
          lean_fields: If True, request a smaller response via fields projection.
  
      Returns:
          A dict with keys: provider, query (sanitized), searchInfo, nextPage,
          latency_ms, results (normalized), raw (subset), trace (q hash).
  
      Raises:
          ValueError: For invalid parameter values (e.g., unsupported safe).
          RuntimeError: For Google API errors or network failures.
      """