search_and_fetch

Instructions

Implementation Reference

• agentfetch/mcp/tools/search_fetch.py:8-33 (handler)

  Core handler for the search_and_fetch tool. Uses Jina's search endpoint to find URLs, then runs each through the AgentFetch fetch_many pipeline. Returns combined results with query, count, and results array.

  def search_and_fetch(
      query: str,
      num_results: int = 3,
      max_tokens_each: int = 2000,
  ) -> dict:
      """Search the web and return clean content from the top results.
  
      Wraps Jina's search endpoint to find URLs, then runs each through the
      AgentFetch pipeline. Returns one combined response.
  
      Args:
          query: Search query string.
          num_results: Number of top results to fetch (default 3, max 10).
          max_tokens_each: Token budget per result.
      """
      num_results = max(1, min(num_results, 10))
      urls = jina_fetcher.search(query, num_results=num_results)
      if not urls:
          return {
              "query": query,
              "count": 0,
              "results": [],
              "error": "search returned no results",
          }
      results = fetch_many(urls, max_tokens_each=max_tokens_each)
      return {"query": query, "count": len(results), "results": results}

• agentfetch/mcp/tools/search_fetch.py:8-33 (schema)

  Input schema defined via function signature (query: str, num_results: int = 3, max_tokens_each: int = 2000). Output schema is the returned dict with keys: query, count, results, and optionally error.

  def search_and_fetch(
      query: str,
      num_results: int = 3,
      max_tokens_each: int = 2000,
  ) -> dict:
      """Search the web and return clean content from the top results.
  
      Wraps Jina's search endpoint to find URLs, then runs each through the
      AgentFetch pipeline. Returns one combined response.
  
      Args:
          query: Search query string.
          num_results: Number of top results to fetch (default 3, max 10).
          max_tokens_each: Token budget per result.
      """
      num_results = max(1, min(num_results, 10))
      urls = jina_fetcher.search(query, num_results=num_results)
      if not urls:
          return {
              "query": query,
              "count": 0,
              "results": [],
              "error": "search returned no results",
          }
      results = fetch_many(urls, max_tokens_each=max_tokens_each)
      return {"query": query, "count": len(results), "results": results}

• agentfetch/mcp/server.py:163-187 (registration)

  Registration of the search_and_fetch tool via the @mcp.tool() decorator. The server function delegates to the imported _search_and_fetch from agentfetch.mcp.tools.search_fetch.

  @mcp.tool()
  def search_and_fetch(
      query: str, num_results: int = 3, max_tokens_each: int = 2000
  ) -> dict:
      """Web search + fetch top results in one call.
  
      WHEN TO USE:
      - You have a research question, not specific URLs. E.g. "what's the
        latest on X", "find docs for Y library", "recent news about Z".
      - You'd otherwise have to call a search tool, parse results, then call
        fetch — this collapses that into one round-trip.
  
      Args:
          query: Search query (2–500 chars).
          num_results: Top N to fetch (1–10, default 3).
          max_tokens_each: Per-result cap (default 2000).
  
      Returns:
          {"query": str, "count": int, "results": [<fetch_url shape>, ...]}
      """
      return _search_and_fetch(
          query=query,
          num_results=num_results,
          max_tokens_each=max_tokens_each,
      )

• agentfetch/mcp/server.py:23-25 (registration)

  Import of the search_and_fetch handler (aliased as _search_and_fetch) from the tools module into the server.

  from agentfetch.mcp.tools.search_fetch import (
      search_and_fetch as _search_and_fetch,
  )

• agentfetch/core/fetchers/jina.py:74-105 (helper)

  Jina search helper — uses Jina's /search/ endpoint to convert a query into a list of URLs. Used by search_and_fetch to discover URLs before fetching them.

  def search(query: str, num_results: int = 3, timeout: int = 30) -> list[str]:
      """Use Jina's `/search/` endpoint to get top-N URLs for a query.
  
      The fetch pipeline is responsible for actually fetching the bodies — this
      function only returns URLs.
      """
      headers = {"Accept": "application/json"}
      api_key = get_settings().jina_api_key
      if api_key:
          headers["Authorization"] = f"Bearer {api_key}"
      try:
          resp = httpx.get(
              f"https://s.jina.ai/{query}",
              headers=headers,
              timeout=timeout,
          )
      except httpx.HTTPError:
          return []
      if resp.status_code != 200:
          return []
      try:
          data = resp.json()
      except ValueError:
          return []
      results = data.get("data") if isinstance(data, dict) else data
      if not isinstance(results, list):
          return []
      urls: list[str] = []
      for item in results[:num_results]:
          if isinstance(item, dict) and "url" in item:
              urls.append(item["url"])
      return urls