cancel_flow
Stop a running forensic investigation flow in the Velociraptor platform by providing client and flow IDs to manage endpoint collection processes.
Instructions
Cancel a running collection flow.
Args: client_id: The client ID (e.g., 'C.1234567890abcdef') flow_id: The flow ID (e.g., 'F.1234567890')
Returns: Cancellation status.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| client_id | Yes | ||
| flow_id | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- The handler implementation for the 'cancel_flow' tool, which performs input validation and executes a VQL query to cancel a Velociraptor flow.
async def cancel_flow( client_id: str, flow_id: str, ) -> list[TextContent]: """Cancel a running collection flow. Args: client_id: The client ID (e.g., 'C.1234567890abcdef') flow_id: The flow ID (e.g., 'F.1234567890') Returns: Cancellation status. """ try: # Input validation client_id = validate_client_id(client_id) flow_id = validate_flow_id(flow_id) client = get_client() vql = f"SELECT cancel_flow(client_id='{client_id}', flow_id='{flow_id}') FROM scope()" results = client.query(vql) return [TextContent( type="text", text=json.dumps({ "client_id": client_id, "flow_id": flow_id, "action": "cancelled", "result": results[0] if results else None, }, indent=2, default=str) )] except grpc.RpcError as e: error_response = map_grpc_error(e, f"cancelling flow {flow_id}") # Check if it's a not-found error if "NOT_FOUND" in error_response.get("grpc_status", ""): error_response["hint"] = f"Flow {flow_id} may not exist for client {client_id}. Use list_flows(client_id='{client_id}') to see available flows." 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 valid client ID (C.*) and flow ID (F.*)" }) )] except Exception: # Generic errors - don't expose internals return [TextContent( type="text", text=json.dumps({ "error": "Failed to cancel flow", "hint": "Check IDs and Velociraptor server connection" }) )] - src/megaraptor_mcp/tools/flows.py:275-275 (registration)Registration of the 'cancel_flow' function as an MCP tool using the @mcp.tool() decorator.
@mcp.tool()