Megaraptor MCP

Instructions

Implementation Reference

• src/megaraptor_mcp/tools/clients.py:22-108 (handler)

  The implementation of the list_clients tool, which queries Velociraptor for clients and returns them as formatted text content.

  @mcp.tool()
  async def list_clients(
      search: Optional[str] = None,
      limit: int = 100,
  ) -> list[TextContent]:
      """Search and list Velociraptor clients (endpoints).
  
      Args:
          search: Optional search query. Supports prefixes like 'label:' and 'host:'.
                 Examples: 'label:production', 'host:workstation-01', 'windows'
          limit: Maximum number of clients to return (default 100)
  
      Returns:
          List of clients with their ID, hostname, OS, labels, and last seen time.
      """
      try:
          # Validate inputs
          limit = validate_limit(limit)
  
          # Basic injection protection for search parameter
          if search and (";" in search or "--" in search):
              return [TextContent(
                  type="text",
                  text=json.dumps({
                      "error": "Invalid search query: potentially unsafe characters detected",
                      "hint": "Remove semicolons and SQL comment markers from search query"
                  })
              )]
  
          client = get_client()
  
          if search:
              vql = f"SELECT * FROM clients(search='{search}') LIMIT {limit}"
          else:
              vql = f"SELECT * FROM clients() LIMIT {limit}"
  
          results = client.query(vql)
  
          # Format the results
          formatted = []
          for row in results:
              client_info = {
                  "client_id": row.get("client_id", ""),
                  "hostname": row.get("os_info", {}).get("hostname", ""),
                  "os": row.get("os_info", {}).get("system", ""),
                  "release": row.get("os_info", {}).get("release", ""),
                  "labels": row.get("labels", []),
                  "last_seen_at": row.get("last_seen_at", ""),
                  "first_seen_at": row.get("first_seen_at", ""),
                  "last_ip": row.get("last_ip", ""),
              }
              formatted.append(client_info)
  
          return [TextContent(
              type="text",
              text=json.dumps(formatted, indent=2)
          )]
  
      except ValueError as e:
          # Validation errors
          return [TextContent(
              type="text",
              text=json.dumps({
                  "error": str(e),
                  "hint": "Check your limit parameter value"
              })
          )]
  
      except grpc.RpcError as e:
          # gRPC errors
          error_info = map_grpc_error(e, "listing clients")
          return [TextContent(
              type="text",
              text=json.dumps(error_info, indent=2)
          )]
  
      except Exception:
          # Generic errors - don't expose internals
          return [TextContent(
              type="text",
              text=json.dumps({
                  "error": "Failed to list clients",
                  "hint": "Check Velociraptor server connection and try again"
              })
          )]