"""Execution Handler - manages n8n workflow executions.""" import asyncio import json from typing import Any from plugins.n8n.client import N8nApiError, N8nClient def _error_json(exc: Exception) -> str: if isinstance(exc, N8nApiError): return json.dumps({"success": False, **exc.to_dict()}, indent=2) return json.dumps({"success": False, "error": str(exc)}, indent=2) def get_tool_specifications() -> list[dict[str, Any]]: """Return tool specifications for ToolGenerator""" return [ # === LIST EXECUTIONS === { "name": "list_executions", "method_name": "list_executions", "description": "List workflow executions with filters by status, workflow, and date. Returns execution history. All filter parameters are OPTIONAL.", "schema": { "type": "object", "properties": { "workflow_id": { "type": "string", "description": "OPTIONAL: Filter by workflow ID. Omit for all executions.", }, "status": { "type": "string", "enum": ["success", "error", "waiting", "running", "new"], "description": "OPTIONAL: Filter by execution status. Omit for all statuses.", }, "project_id": { "type": "string", "description": "OPTIONAL: Filter by project ID (Enterprise). Omit for all projects.", }, "include_data": { "type": "boolean", "description": "Include full execution data", "default": False, }, "limit": { "type": "integer", "description": "Maximum results", "default": 20, "minimum": 1, "maximum": 250, }, "cursor": {"type": "string", "description": "OPTIONAL: Pagination cursor."}, }, }, "scope": "read", }, # === GET EXECUTION === { "name": "get_execution", "method_name": "get_execution", "description": "Get detailed information about a specific execution including status, timing, and node outputs.", "schema": { "type": "object", "properties": { "execution_id": { "type": "string", "description": "Execution ID", "minLength": 1, }, "include_data": { "type": "boolean", "description": "Include full node execution data", "default": True, }, }, "required": ["execution_id"], }, "scope": "read", }, # === DELETE EXECUTION === { "name": "delete_execution", "method_name": "delete_execution", "description": "Delete a single execution record. This removes the execution from history.", "schema": { "type": "object", "properties": { "execution_id": { "type": "string", "description": "Execution ID to delete", "minLength": 1, } }, "required": ["execution_id"], }, "scope": "write", }, # === DELETE EXECUTIONS (BULK) === { "name": "delete_executions", "method_name": "delete_executions", "description": "Bulk delete multiple execution records.", "schema": { "type": "object", "properties": { "execution_ids": { "type": "array", "description": "List of execution IDs to delete", "items": {"type": "string"}, "minItems": 1, } }, "required": ["execution_ids"], }, "scope": "write", }, # === STOP EXECUTION === { "name": "stop_execution", "method_name": "stop_execution", "description": "Stop a currently running execution.", "schema": { "type": "object", "properties": { "execution_id": { "type": "string", "description": "Execution ID to stop", "minLength": 1, } }, "required": ["execution_id"], }, "scope": "write", }, # === RETRY EXECUTION === { "name": "retry_execution", "method_name": "retry_execution", "description": "Retry a failed execution. Creates a new execution with the same input data.", "schema": { "type": "object", "properties": { "execution_id": { "type": "string", "description": "Execution ID to retry", "minLength": 1, } }, "required": ["execution_id"], }, "scope": "write", }, # === GET EXECUTION DATA === { "name": "get_execution_data", "method_name": "get_execution_data", "description": "Get full execution data including all node inputs and outputs. Useful for debugging.", "schema": { "type": "object", "properties": { "execution_id": { "type": "string", "description": "Execution ID", "minLength": 1, } }, "required": ["execution_id"], }, "scope": "read", }, # === WAIT FOR EXECUTION === { "name": "wait_for_execution", "method_name": "wait_for_execution", "description": "Poll an execution until it completes. Returns final status and output.", "schema": { "type": "object", "properties": { "execution_id": { "type": "string", "description": "Execution ID to wait for", "minLength": 1, }, "timeout_seconds": { "type": "integer", "description": "Maximum seconds to wait", "default": 60, "minimum": 5, "maximum": 300, }, "poll_interval": { "type": "integer", "description": "Seconds between status checks", "default": 2, "minimum": 1, "maximum": 30, }, }, "required": ["execution_id"], }, "scope": "read", }, ] # === HANDLER FUNCTIONS === async def list_executions( client: N8nClient, workflow_id: str | None = None, status: str | None = None, project_id: str | None = None, include_data: bool = False, limit: int = 20, cursor: str | None = None, ) -> str: """List workflow executions.""" try: response = await client.list_executions( workflow_id=workflow_id, status=status, project_id=project_id, include_data=include_data, limit=limit, cursor=cursor, ) executions = response.get("data", []) next_cursor = response.get("nextCursor") result = { "success": True, "count": len(executions), "executions": [ { "id": e.get("id"), "workflow_id": e.get("workflowId"), "workflow_name": e.get("workflowData", {}).get("name"), "status": e.get("status"), "finished": e.get("finished"), "started_at": e.get("startedAt"), "stopped_at": e.get("stoppedAt"), "mode": e.get("mode"), } for e in executions ], "next_cursor": next_cursor, } return json.dumps(result, indent=2) except Exception as e: return _error_json(e) async def get_execution(client: N8nClient, execution_id: str, include_data: bool = True) -> str: """Get execution details""" try: execution = await client.get_execution(execution_id, include_data) result = { "success": True, "execution": { "id": execution.get("id"), "workflow_id": execution.get("workflowId"), "workflow_name": execution.get("workflowData", {}).get("name"), "status": execution.get("status"), "finished": execution.get("finished"), "started_at": execution.get("startedAt"), "stopped_at": execution.get("stoppedAt"), "mode": execution.get("mode"), "retry_of": execution.get("retryOf"), "retry_success_id": execution.get("retrySuccessId"), }, } # Include node execution data if requested if include_data and "data" in execution: result["execution"]["node_data"] = execution.get("data") return json.dumps(result, indent=2) except Exception as e: return _error_json(e) async def delete_execution(client: N8nClient, execution_id: str) -> str: """Delete a single execution""" try: await client.delete_execution(execution_id) result = {"success": True, "message": f"Execution {execution_id} deleted successfully"} return json.dumps(result, indent=2) except Exception as e: return _error_json(e) async def delete_executions(client: N8nClient, execution_ids: list[str]) -> str: """Bulk delete executions""" try: deleted = [] failed = [] for exec_id in execution_ids: try: await client.delete_execution(exec_id) deleted.append(exec_id) except Exception as e: failed.append({"id": exec_id, "error": str(e)}) result = { "success": len(failed) == 0, "deleted_count": len(deleted), "deleted": deleted, "failed": failed if failed else None, } return json.dumps(result, indent=2) except Exception as e: return _error_json(e) async def stop_execution(client: N8nClient, execution_id: str) -> str: """Stop a running execution""" try: await client.stop_execution(execution_id) result = {"success": True, "message": f"Execution {execution_id} stopped successfully"} return json.dumps(result, indent=2) except Exception as e: return _error_json(e) async def retry_execution(client: N8nClient, execution_id: str) -> str: """Retry a failed execution""" try: # Get the original execution to find workflow ID original = await client.get_execution(execution_id, include_data=True) workflow_id = original.get("workflowId") if not workflow_id: raise Exception("Cannot find workflow ID from execution") # Get input data from original execution input_data = None exec_data = original.get("data", {}) if exec_data: # Try to get input from first node result_data = exec_data.get("resultData", {}) run_data = result_data.get("runData", {}) if run_data: first_node = list(run_data.keys())[0] if run_data else None if first_node: node_data = run_data[first_node] if node_data and len(node_data) > 0: input_data = node_data[0].get("data", {}).get("main", [[]])[0] # Execute the workflow again new_execution = await client.execute_workflow(workflow_id, data=input_data) result = { "success": True, "message": f"Retrying execution {execution_id}", "original_execution_id": execution_id, "new_execution": { "id": new_execution.get("id"), "workflow_id": workflow_id, "status": new_execution.get("status", "running"), }, } return json.dumps(result, indent=2) except Exception as e: return _error_json(e) async def get_execution_data(client: N8nClient, execution_id: str) -> str: """Get full execution data""" try: execution = await client.get_execution(execution_id, include_data=True) exec_data = execution.get("data", {}) result_data = exec_data.get("resultData", {}) result = { "success": True, "execution_id": execution_id, "status": execution.get("status"), "finished": execution.get("finished"), "run_data": result_data.get("runData", {}), "last_node_executed": result_data.get("lastNodeExecuted"), "error": result_data.get("error"), } return json.dumps(result, indent=2) except Exception as e: return _error_json(e) async def wait_for_execution( client: N8nClient, execution_id: str, timeout_seconds: int = 60, poll_interval: int = 2 ) -> str: """Wait for execution to complete""" try: elapsed = 0 final_status = None execution = None while elapsed < timeout_seconds: execution = await client.get_execution(execution_id, include_data=True) status = execution.get("status") finished = execution.get("finished") if finished or status in ["success", "error", "crashed"]: final_status = status break await asyncio.sleep(poll_interval) elapsed += poll_interval if not final_status: return json.dumps( { "success": False, "error": f"Timeout after {timeout_seconds} seconds", "execution_id": execution_id, "last_status": execution.get("status") if execution else "unknown", }, indent=2, ) result = { "success": True, "execution_id": execution_id, "final_status": final_status, "finished": execution.get("finished"), "started_at": execution.get("startedAt"), "stopped_at": execution.get("stoppedAt"), "duration_seconds": elapsed, } # Include error if failed if final_status == "error": exec_data = execution.get("data", {}) result_data = exec_data.get("resultData", {}) result["error"] = result_data.get("error") return json.dumps(result, indent=2) except Exception as e: return _error_json(e)