OpenTelemetry MCP Server

Instructions

Implementation Reference

• src/tools/loki_tools.py:58-125 (handler)

  Implementation of the search_logs tool handler in src/tools/loki_tools.py.

  async def search_logs(
      client: LokiClient,
      service: Optional[str] = None,
      namespace: Optional[str] = None,
      search_text: Optional[str] = None,
      level: Optional[str] = None,
      start: Optional[str] = None,
      end: Optional[str] = None,
      limit: int = 100
  ) -> Dict[str, Any]:
      """
      Search logs with simple filters.
      
      Args:
          client: Loki client
          service: Service/job name filter
          namespace: Namespace filter
          search_text: Text to search for in logs
          level: Log level filter (error, warn, info, debug)
          start: Start time
          end: End time
          limit: Maximum number of results
          
      Returns:
          Filtered log entries
      """
      try:
          # Build LogQL query
          labels = []
          if service:
              labels.append(f'job="{service}"')
          if namespace:
              labels.append(f'namespace="{namespace}"')
          
          query = "{" + ",".join(labels) + "}" if labels else "{job=~\".+\"}"
          
          # Add search text filter
          if search_text:
              query += f' |= "{search_text}"'
          
          # Add level filter
          if level:
              query += f' | json | level="{level}"'
          
          # Parse time range
          start_dt, end_dt = parse_time_range(start, end)
          start_ns = to_loki_time(start_dt)
          end_ns = to_loki_time(end_dt)
          
          result = await client.query_range(query, start_ns, end_ns, limit)
          
          return {
              "success": True,
              "query": query,
              "filters": {
                  "service": service,
                  "namespace": namespace,
                  "search_text": search_text,
                  "level": level
              },
              "result": result
          }
      except Exception as e:
          logger.error(f"Error searching logs: {e}")
          return {
              "success": False,
              "error": str(e)
          }