Uptrace MCP Server

Instructions

Implementation Reference

• src/uptrace_mcp/server.py:631-696 (handler)

  The handler function logic that processes arguments and fetches logs from the Uptrace client.

  elif name == "uptrace_search_logs":
      hours = arguments.get("hours", 3)
      search_text = arguments.get("search_text")
      severity = arguments.get("severity")
      service_name = arguments.get("service_name")
      query = arguments.get("query")
      limit = arguments.get("limit", 100)
  
      time_lt = datetime.now(timezone.utc)
      time_gte = time_lt - timedelta(hours=hours)
  
      logger.info(
          "Searching logs: text=%s, severity=%s, service=%s (limit: %s)",
          search_text,
          severity,
          service_name,
          limit,
      )
  
      # Build UQL query for log search
      log_query_parts = []
      if search_text:
          # Search in event attribute (which contains log message)
          log_query_parts.append(f'where event contains "{search_text}"')
      if severity:
          log_query_parts.append(f'where log_severity = "{severity}"')
      if service_name:
          log_query_parts.append(f'where service_name = "{service_name}"')
      if query:
          log_query_parts.append(query)
  
      # Use spans API to query logs (logs are represented as spans)
      full_query = 'where _system = "log:all"'
      if log_query_parts:
          full_query += " | " + " | ".join(log_query_parts)
  
      response = client.get_spans(
          time_gte=time_gte,
          time_lt=time_lt,
          query=full_query,
          limit=limit,
      )
  
      lines = [
          "# Logs Search Results",
          f"**Time Range**: {time_gte.isoformat()} - {time_lt.isoformat()}",
          f"**Total Logs**: {response.count}",
          f"**Returned**: {len(response.spans)}",
          "",
      ]
  
      if search_text:
          lines.append(f"**Search Text**: `{search_text}`")
      if severity:
          lines.append(f"**Severity**: {severity}")
      if service_name:
          lines.append(f"**Service**: {service_name}")
      lines.append("")
  
      if response.spans:
          # Group by severity
          by_severity: Dict[str, int] = {}
          by_service: Dict[str, int] = {}
          for span in response.spans:
              sev = span.attrs.get("log_severity", "UNKNOWN")
              service = span.attrs.get("service_name", "unknown")

• src/uptrace_mcp/server.py:264-292 (schema)

  The tool definition, including the name, description, and input schema.

  Tool(
      name="uptrace_search_logs",
      description="Search logs by text, severity, service name, or custom UQL query. Logs are represented as spans with _system = 'log:all'.",
      inputSchema={
          "type": "object",
          "properties": {
              "hours": {
                  "type": "integer",
                  "description": "Number of hours to look back (default: 3)",
                  "default": 3,
              },
              "search_text": {
                  "type": "string",
                  "description": "Text to search for in log messages (case-insensitive)",
              },
              "severity": {
                  "type": "string",
                  "description": "Filter by log severity (e.g., 'ERROR', 'WARN', 'INFO', 'DEBUG')",
                  "enum": ["DEBUG", "INFO", "WARN", "ERROR", "FATAL"],
              },
              "service_name": {
                  "type": "string",
                  "description": "Filter by service name",
              },
              "query": {
                  "type": "string",
                  "description": "Additional UQL query string for advanced filtering",
              },
              "limit": {