kagi_search_fetch

Instructions

Implementation Reference

• src/kagi_session_mcp/server.py:81-117 (handler)

  The main handler function for the 'kagi_search_fetch' MCP tool. Decorated with @mcp.tool(), it accepts a list of search queries and an optional limit, executes them concurrently via asyncio.gather, and formats results using format_search_results.

  @mcp.tool()
  async def kagi_search_fetch(
      queries: list[str] = Field(
          description=(
              "One or more concise, keyword-focused search queries. "
              "Include essential context within each query for standalone use. "
              "Supports Kagi operators: site:, -site:, filetype:/ext:, intitle:, "
              'inurl:, lang:, loc:, before:, after:, "exact phrase", +term, -term'
          )
      ),
      limit: int | None = Field(
          default=None,
          description="Maximum number of results per query. Default: all available.",
      ),
  ) -> str:
      """Fetch web results based on one or more queries using Kagi Search.
  
      Use for general search and when the user explicitly tells you to
      'fetch' results/information. Results are from all queries given.
      They are numbered continuously, so that a user may be able to refer
      to a result by a specific number.
      """
      if not queries:
          raise ValueError("Search called with no queries.")
  
      if client is None:
          raise RuntimeError("Server not initialized. Session client is missing.")
  
      # Execute all queries concurrently using asyncio.gather for parallel I/O
      query_response_pairs = await asyncio.gather(
          *[_execute_single_query(query, limit) for query in queries]
      )
  
      # Preserve original query ordering for consistent result numbering
      responses = [response for _, response in query_response_pairs]
  
      return format_search_results(queries, responses)

• src/kagi_session_mcp/server.py:81-94 (schema)

  Pydantic Field definitions for the 'kagi_search_fetch' tool parameters: 'queries' (list of keyword search strings with Kagi operators) and 'limit' (max results per query).

  @mcp.tool()
  async def kagi_search_fetch(
      queries: list[str] = Field(
          description=(
              "One or more concise, keyword-focused search queries. "
              "Include essential context within each query for standalone use. "
              "Supports Kagi operators: site:, -site:, filetype:/ext:, intitle:, "
              'inurl:, lang:, loc:, before:, after:, "exact phrase", +term, -term'
          )
      ),
      limit: int | None = Field(
          default=None,
          description="Maximum number of results per query. Default: all available.",
      ),

• src/kagi_session_mcp/server.py:81-81 (registration)

  Tool registration via the @mcp.tool() decorator on the kagi_search_fetch async function, which registers it with the FastMCP server instance.

  @mcp.tool()

• src/kagi_session_mcp/server.py:55-78 (helper)

  Helper function _execute_single_query that executes a single search query: fetches HTML via client.search_html(), parses it via parse_search_html(), and applies the limit. Used by kagi_search_fetch in the asyncio.gather call.

  async def _execute_single_query(
      query: str, limit: int | None
  ) -> tuple[str, "SearchResponse"]:
      """Execute a single search query and return (query, response) tuple.
  
      Runs the HTML fetch and parse pipeline with timing for elapsed_ms.
  
      Args:
          query: Search query string
          limit: Maximum number of results to return, or None for all
  
      Returns:
          Tuple of (original query, SearchResponse)
      """
      start = time.monotonic()
      html = await client.search_html(query)
      elapsed_ms = int((time.monotonic() - start) * 1000)
      response = parse_search_html(html, query, elapsed_ms)
  
      # Apply limit if specified
      if limit is not None and limit > 0:
          response.data = response.data[:limit]
  
      return (query, response)

• src/kagi_session_mcp/formatter.py:10-53 (helper)

  format_search_results helper that formats SearchResponse objects into numbered text output, used by kagi_search_fetch to produce the final string result.

  def format_search_results(
      queries: list[str], responses: list[SearchResponse]
  ) -> str:
      """Format search results as numbered text, matching official kagimcp output format.
  
      Results from all queries are numbered continuously so that users can
      refer to a specific result by its number.
  
      Args:
          queries: List of search queries
          responses: List of SearchResponse objects (one per query)
  
      Returns:
          Formatted text string with numbered results
      """
      if not queries or not responses:
          return "No search results found."
  
      parts: list[str] = []
      result_number = 0
  
      for query_idx, (query, response) in enumerate(zip(queries, responses)):
          # Add query header
          if len(queries) > 1:
              parts.append(f"## Query: {query}\n")
          else:
              parts.append(f"## Search results for: {query}\n")
  
          if not response.data:
              parts.append("No results found for this query.\n")
              continue
  
          for item in response.data:
              if isinstance(item, SearchResult):
                  result_number += 1
                  parts.append(_format_search_result(item, result_number))
              elif isinstance(item, RelatedSearches):
                  parts.append(_format_related_searches(item))
  
          # Add separator between queries
          if query_idx < len(queries) - 1:
              parts.append("---\n")
  
      return "\n".join(parts)