search_documentation

Instructions

Implementation Reference

• opentelemetry_documentation_mcp_server/server.py:187-305 (handler)

  The main execution logic for the 'search_documentation' tool. This async function handles the Google Custom Search API call to opentelemetry.io documentation, processes results using parse_search_results, and returns a list of SearchResult objects. The @mcp.tool() decorator registers it as an MCP tool with input schema from Pydantic Fields.

  @mcp.tool()
  async def search_documentation(
      ctx: Context,
      search_phrase: str = Field(description='Search phrase to use'),
      limit: int = Field(
          default=10,
          description='Maximum number of results to return',
          ge=1,
          le=50,
      ),
  ) -> List[SearchResult]:
      """Search OpenTelemetry documentation using Google Custom Search.
  
      ## Usage
  
      This tool searches across OpenTelemetry documentation for pages matching your search phrase.
      Use it to find relevant documentation when you don't have a specific URL.
  
      ## Search Tips
  
      - Use specific technical terms rather than general phrases
      - Include OpenTelemetry concepts to narrow results (e.g., "tracing instrumentation" instead of just "tracing")
      - Use quotes for exact phrase matching (e.g., "SDK configuration")
      - Include abbreviations and alternative terms to improve results (e.g., "OTEL collector")
  
      ## API Limits
  
      The search uses Google's Custom Search API which has usage limits:
      - Free tier: 100 queries per day
      - Results are limited to 10 per page
  
      ## Result Interpretation
  
      Each result includes:
      - rank_order: The relevance ranking (lower is more relevant)
      - url: The documentation page URL
      - title: The page title
      - context: A brief excerpt or summary (if available)
  
      Args:
          ctx: MCP context for logging and error handling
          search_phrase: Search phrase to use
          limit: Maximum number of results to return (will be capped at 10 due to API limitations)
  
      Returns:
          List of search results with URLs, titles, and context snippets
      """
      logger.debug(f'Searching OpenTelemetry documentation for: {search_phrase}')
  
      # Get API key from environment variable
      api_key = os.getenv('GOOGLE_API_KEY')
      if not api_key:
          error_msg = (
              'Google API key not found. To use the search_documentation tool, you need to set the '
              'GOOGLE_API_KEY environment variable. You can get an API key from the Google Cloud Console '
              '(https://console.cloud.google.com/apis/credentials) and add it to your MCP configuration.'
          )
          logger.error(error_msg)
          await ctx.error(error_msg)
          return [SearchResult(rank_order=1, url='', title=error_msg, context=None)]
  
      # Cap limit at 10 due to Google CSE API constraints
      if limit > 10:
          logger.warning(f'Limiting search results to 10 (API limit), requested: {limit}')
          limit = 10
  
      params = {
          'key': api_key,
          'cx': OPENTELEMETRY_SEARCH_CX,
          'q': search_phrase,
          'num': limit,
      }
  
      async with httpx.AsyncClient() as client:
          try:
              response = await client.get(
                  GOOGLE_SEARCH_API_URL,
                  params=params,
                  headers={'User-Agent': DEFAULT_USER_AGENT},
                  timeout=30,
              )
          except httpx.HTTPError as e:
              error_msg = f'Error searching OpenTelemetry docs: {str(e)}'
              logger.error(error_msg)
              await ctx.error(error_msg)
              return [SearchResult(rank_order=1, url='', title=error_msg, context=None)]
  
          if response.status_code >= 400:
              error_msg = f'Error searching OpenTelemetry docs - status code {response.status_code}'
              logger.error(error_msg)
              await ctx.error(error_msg)
              return [
                  SearchResult(
                      rank_order=1,
                      url='',
                      title=error_msg,
                      context=None,
                  )
              ]
  
          try:
              data = response.json()
          except json.JSONDecodeError as e:
              error_msg = f'Error parsing search results: {str(e)}'
              logger.error(error_msg)
              await ctx.error(error_msg)
              return [
                  SearchResult(
                      rank_order=1,
                      url='',
                      title=error_msg,
                      context=None,
                  )
              ]
  
          results = parse_search_results(data)
  
      logger.debug(f'Found {len(results)} search results for: {search_phrase}')
      return results

• opentelemetry_documentation_mcp_server/models.py:21-28 (schema)

  Pydantic BaseModel defining the structure of each search result returned by the search_documentation tool: rank_order, url, title, and optional context snippet.

  class SearchResult(BaseModel):
      """Search result from OpenTelemetry documentation search."""
  
      rank_order: int
      url: str
      title: str
      context: Optional[str] = None

• opentelemetry_documentation_mcp_server/server.py:187-187 (registration)

  The @mcp.tool() decorator registers the search_documentation function as an MCP tool.

  @mcp.tool()

• opentelemetry_documentation_mcp_server/util.py:182-205 (helper)

  Helper function that parses the raw Google Custom Search JSON response into a list of SearchResult models, used by the search_documentation handler.

  def parse_search_results(data: Dict[str, Any]) -> List[SearchResult]:
      """Parse Google Custom Search results into structured format.
  
      Args:
          data: Raw API response data from Google Custom Search
  
      Returns:
          List of SearchResult objects in a standard format
      """
      results = []
  
      if 'items' in data:
          for i, item in enumerate(data['items']):
              results.append(
                  SearchResult(
                      rank_order=i + 1,
                      url=item.get('link', ''),
                      title=item.get('title', ''),
                      context=item.get('snippet'),
                  )
              )
  
      return results