Megaraptor MCP

Instructions

Implementation Reference

• src/megaraptor_mcp/tools/hunts.py:305-390 (handler)

  The handler implementation for the modify_hunt tool, which updates the state of a Velociraptor hunt by executing a VQL query.

  async def modify_hunt(
      hunt_id: str,
      action: str,
  ) -> list[TextContent]:
      """Modify a Velociraptor hunt state.
  
      Args:
          hunt_id: The hunt ID (e.g., 'H.1234567890')
          action: Action to perform: 'start', 'pause', 'stop', 'archive'
  
      Returns:
          Updated hunt status.
      """
      try:
          # Input validation
          hunt_id = validate_hunt_id(hunt_id)
  
          action_map = {
              "start": "StartHuntRequest",
              "pause": "PauseHuntRequest",
              "stop": "StopHuntRequest",
              "archive": "ArchiveHuntRequest",
          }
  
          if action not in action_map:
              return [TextContent(
                  type="text",
                  text=json.dumps({
                      "error": f"Invalid action '{action}'. Must be one of: start, pause, stop, archive"
                  })
              )]
  
          client = get_client()
  
          # Use the hunt() function to modify the hunt
          if action == "start":
              vql = f"SELECT hunt_update(hunt_id='{hunt_id}', state='RUNNING') FROM scope()"
          elif action == "pause":
              vql = f"SELECT hunt_update(hunt_id='{hunt_id}', state='PAUSED') FROM scope()"
          elif action == "stop":
              vql = f"SELECT hunt_update(hunt_id='{hunt_id}', state='STOPPED') FROM scope()"
          else:  # archive
              vql = f"SELECT hunt_update(hunt_id='{hunt_id}', state='ARCHIVED') FROM scope()"
  
          results = client.query(vql)
  
          return [TextContent(
              type="text",
              text=json.dumps({
                  "hunt_id": hunt_id,
                  "action": action,
                  "status": "success",
                  "result": results[0] if results else None,
              }, indent=2, default=str)
          )]
  
      except grpc.RpcError as e:
          error_response = map_grpc_error(e, f"modifying hunt {hunt_id}")
          # Check if it's a not-found error
          if "NOT_FOUND" in error_response.get("grpc_status", ""):
              error_response["hint"] = f"Hunt {hunt_id} may not exist. Use list_hunts() to see available hunts."
          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": "Provide a valid hunt ID starting with 'H.'"
              })
          )]
  
      except Exception:
          # Generic errors - don't expose internals
          return [TextContent(
              type="text",
              text=json.dumps({
                  "error": "Failed to modify hunt",
                  "hint": "Check hunt ID and action parameter"
              })
          )]