Megaraptor MCP

Instructions

Implementation Reference

• src/megaraptor_mcp/tools/hunts.py:148-230 (handler)

  The handler implementation for the list_hunts tool which queries Velociraptor hunts via VQL and formats the output.

  async def list_hunts(
      state: Optional[str] = None,
      limit: int = 50,
  ) -> list[TextContent]:
      """List Velociraptor hunts.
  
      Args:
          state: Optional filter by state: 'RUNNING', 'PAUSED', 'STOPPED', 'COMPLETED'
          limit: Maximum number of hunts to return (default 50)
  
      Returns:
          List of hunts with their status and statistics.
      """
      try:
          # Input validation
          limit = validate_limit(limit)
  
          if state and state.upper() not in ['RUNNING', 'PAUSED', 'STOPPED', 'COMPLETED']:
              return [TextContent(
                  type="text",
                  text=json.dumps({
                      "error": f"Invalid state '{state}'. Must be one of: RUNNING, PAUSED, STOPPED, COMPLETED"
                  })
              )]
          client = get_client()
  
          vql = f"SELECT * FROM hunts() LIMIT {limit}"
          results = client.query(vql)
  
          # Filter by state if specified
          if state:
              results = [r for r in results if r.get("state", "").upper() == state.upper()]
  
          # Format the results
          formatted = []
          for row in results:
              hunt = {
                  "hunt_id": row.get("hunt_id", ""),
                  "description": row.get("hunt_description", ""),
                  "state": row.get("state", ""),
                  "artifacts": row.get("artifacts", []),
                  "created_time": row.get("create_time", ""),
                  "start_time": row.get("start_time", ""),
                  "stats": {
                      "total_clients_scheduled": row.get("stats", {}).get("total_clients_scheduled", 0),
                      "total_clients_with_results": row.get("stats", {}).get("total_clients_with_results", 0),
                      "total_clients_with_errors": row.get("stats", {}).get("total_clients_with_errors", 0),
                  },
                  "creator": row.get("creator", ""),
              }
              formatted.append(hunt)
  
          return [TextContent(
              type="text",
              text=json.dumps(formatted, indent=2, default=str)
          )]
  
      except grpc.RpcError as e:
          error_response = map_grpc_error(e, "hunt listing")
          return [TextContent(
              type="text",
              text=json.dumps(error_response)
          )]
  
      except ValueError as e:
          # Validation errors
          return [TextContent(
              type="text",
              text=json.dumps({
                  "error": str(e),
                  "hint": "Check your limit parameter value"
              })
          )]
  
      except Exception:
          # Generic errors - don't expose internals
          return [TextContent(
              type="text",
              text=json.dumps({
                  "error": "Failed to list hunts",
                  "hint": "Check Velociraptor server connection and try again"
              })
          )]