"""Workflow Handler - manages n8n workflows""" import json from typing import Any from plugins.n8n.client import N8nClient def get_tool_specifications() -> list[dict[str, Any]]: """Return tool specifications for ToolGenerator""" return [ # === LIST WORKFLOWS === { "name": "list_workflows", "method_name": "list_workflows", "description": "List all n8n workflows with optional filters. Returns workflow ID, name, active status, tags, and metadata. All filter parameters are OPTIONAL.", "schema": { "type": "object", "properties": { "active": { "type": "boolean", "description": "OPTIONAL: Filter by active/inactive status. Omit for all workflows.", }, "tags": { "type": "string", "description": "OPTIONAL: Filter by tag name(s), comma-separated. Omit for all workflows.", }, "name": { "type": "string", "description": "OPTIONAL: Filter by workflow name (partial match). Omit for all workflows.", }, "limit": { "type": "integer", "description": "Maximum workflows to return", "default": 50, "minimum": 1, "maximum": 250, }, "cursor": { "type": "string", "description": "OPTIONAL: Pagination cursor for next page.", }, }, }, "scope": "read", }, # === GET WORKFLOW === { "name": "get_workflow", "method_name": "get_workflow", "description": "Get detailed information about a specific workflow including nodes, connections, and settings.", "schema": { "type": "object", "properties": { "workflow_id": {"type": "string", "description": "Workflow ID", "minLength": 1} }, "required": ["workflow_id"], }, "scope": "read", }, # === CREATE WORKFLOW === { "name": "create_workflow", "method_name": "create_workflow", "description": "Create a new workflow from JSON definition. Workflow will be inactive by default. Note: settings and static_data are OPTIONAL - omit them entirely if not needed.", "schema": { "type": "object", "properties": { "name": {"type": "string", "description": "Workflow name", "minLength": 1}, "nodes": { "type": "array", "description": "Array of node definitions", "items": {"type": "object"}, "default": [], }, "connections": { "type": "object", "description": "Node connections definition", "default": {}, }, "settings": { "type": "object", "description": "OPTIONAL: Workflow settings (timezone, error workflow, etc.). Omit this parameter if not needed.", "default": {}, }, "static_data": { "type": "object", "description": "OPTIONAL: Static data for the workflow. Omit this parameter if not needed.", "default": {}, }, }, "required": ["name"], }, "scope": "write", }, # === UPDATE WORKFLOW === { "name": "update_workflow", "method_name": "update_workflow", "description": "Update an existing workflow. Can modify nodes, connections, settings, or name. All parameters except workflow_id are OPTIONAL - omit them to keep current values.", "schema": { "type": "object", "properties": { "workflow_id": { "type": "string", "description": "Workflow ID to update", "minLength": 1, }, "name": { "type": "string", "description": "OPTIONAL: New workflow name. Omit to keep current.", }, "nodes": { "type": "array", "items": {"type": "object"}, "description": "OPTIONAL: Updated node definitions. Omit to keep current.", }, "connections": { "type": "object", "description": "OPTIONAL: Updated connections. Omit to keep current.", }, "settings": { "type": "object", "description": "OPTIONAL: Updated settings. Omit to keep current.", }, }, "required": ["workflow_id"], }, "scope": "write", }, # === DELETE WORKFLOW === { "name": "delete_workflow", "method_name": "delete_workflow", "description": "Permanently delete a workflow. This action cannot be undone.", "schema": { "type": "object", "properties": { "workflow_id": { "type": "string", "description": "Workflow ID to delete", "minLength": 1, } }, "required": ["workflow_id"], }, "scope": "admin", }, # === ACTIVATE WORKFLOW === { "name": "activate_workflow", "method_name": "activate_workflow", "description": "Activate a workflow so it runs automatically when triggered. Workflow must have at least one trigger node.", "schema": { "type": "object", "properties": { "workflow_id": { "type": "string", "description": "Workflow ID to activate", "minLength": 1, } }, "required": ["workflow_id"], }, "scope": "write", }, # === DEACTIVATE WORKFLOW === { "name": "deactivate_workflow", "method_name": "deactivate_workflow", "description": "Deactivate a workflow. It will no longer run automatically but can still be executed manually.", "schema": { "type": "object", "properties": { "workflow_id": { "type": "string", "description": "Workflow ID to deactivate", "minLength": 1, } }, "required": ["workflow_id"], }, "scope": "write", }, # === EXECUTE WORKFLOW === { "name": "execute_workflow", "method_name": "execute_workflow", "description": "Manually execute a workflow and return execution ID. Use get_execution to check status and results.", "schema": { "type": "object", "properties": { "workflow_id": { "type": "string", "description": "Workflow ID to execute", "minLength": 1, } }, "required": ["workflow_id"], }, "scope": "write", }, # === EXECUTE WORKFLOW WITH DATA === { "name": "execute_workflow_with_data", "method_name": "execute_workflow_with_data", "description": "Execute workflow with custom input data. Useful for workflows with webhook or manual triggers.", "schema": { "type": "object", "properties": { "workflow_id": { "type": "string", "description": "Workflow ID to execute", "minLength": 1, }, "data": { "type": "object", "description": "Input data to pass to workflow trigger node", }, }, "required": ["workflow_id", "data"], }, "scope": "write", }, # === DUPLICATE WORKFLOW === { "name": "duplicate_workflow", "method_name": "duplicate_workflow", "description": "Create a copy of an existing workflow with a new name.", "schema": { "type": "object", "properties": { "workflow_id": { "type": "string", "description": "Workflow ID to duplicate", "minLength": 1, }, "new_name": { "type": "string", "description": "Name for the duplicated workflow", "minLength": 1, }, }, "required": ["workflow_id", "new_name"], }, "scope": "write", }, # === EXPORT WORKFLOW === { "name": "export_workflow", "method_name": "export_workflow", "description": "Export a workflow as JSON. Can be used for backup or sharing.", "schema": { "type": "object", "properties": { "workflow_id": { "type": "string", "description": "Workflow ID to export", "minLength": 1, } }, "required": ["workflow_id"], }, "scope": "read", }, # === IMPORT WORKFLOW === { "name": "import_workflow", "method_name": "import_workflow", "description": "Import a workflow from JSON definition. Similar to create but accepts full workflow JSON.", "schema": { "type": "object", "properties": { "workflow_json": { "type": "object", "description": "Full workflow JSON to import", }, "name_override": { "type": "string", "description": "OPTIONAL: Override the workflow name from JSON. Omit to use the name from JSON.", }, }, "required": ["workflow_json"], }, "scope": "write", }, # === GET WORKFLOW TAGS === { "name": "get_workflow_tags", "method_name": "get_workflow_tags", "description": "Get list of tags assigned to a workflow.", "schema": { "type": "object", "properties": { "workflow_id": {"type": "string", "description": "Workflow ID", "minLength": 1} }, "required": ["workflow_id"], }, "scope": "read", }, # === SET WORKFLOW TAGS === { "name": "set_workflow_tags", "method_name": "set_workflow_tags", "description": "Assign tags to a workflow. Replaces existing tags. Pass empty array [] to remove all tags.", "schema": { "type": "object", "properties": { "workflow_id": {"type": "string", "description": "Workflow ID", "minLength": 1}, "tag_ids": { "type": "array", "description": "List of tag IDs to assign. Use empty array [] to remove all tags. Get tag IDs using n8n_list_tags first.", "items": {"type": "string"}, "default": [], }, }, "required": ["workflow_id"], }, "scope": "write", }, ] # === HANDLER FUNCTIONS === async def list_workflows( client: N8nClient, active: bool | None = None, tags: str | None = None, name: str | None = None, limit: int = 50, cursor: str | None = None, ) -> str: """List all workflows with filters""" try: response = await client.list_workflows( active=active, tags=tags, name=name, limit=limit, cursor=cursor ) workflows = response.get("data", []) next_cursor = response.get("nextCursor") result = { "success": True, "count": len(workflows), "workflows": [ { "id": w.get("id"), "name": w.get("name"), "active": w.get("active"), "tags": [t.get("name") for t in w.get("tags", [])], "created_at": w.get("createdAt"), "updated_at": w.get("updatedAt"), } for w in workflows ], "next_cursor": next_cursor, } return json.dumps(result, indent=2) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def get_workflow(client: N8nClient, workflow_id: str) -> str: """Get workflow details""" try: workflow = await client.get_workflow(workflow_id) result = { "success": True, "workflow": { "id": workflow.get("id"), "name": workflow.get("name"), "active": workflow.get("active"), "nodes": workflow.get("nodes", []), "connections": workflow.get("connections", {}), "settings": workflow.get("settings", {}), "static_data": workflow.get("staticData"), "tags": [t.get("name") for t in workflow.get("tags", [])], "created_at": workflow.get("createdAt"), "updated_at": workflow.get("updatedAt"), }, } return json.dumps(result, indent=2) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def create_workflow( client: N8nClient, name: str, nodes: list[dict[str, Any]] | None = None, connections: dict[str, Any] | None = None, settings: dict[str, Any] | None = None, static_data: dict[str, Any] | None = None, ) -> str: """Create a new workflow""" try: # Use defaults for optional parameters data = { "name": name, "nodes": nodes if nodes is not None else [], "connections": connections if connections is not None else {}, } # Only add settings/static_data if provided and non-empty if settings and isinstance(settings, dict) and len(settings) > 0: data["settings"] = settings if static_data and isinstance(static_data, dict) and len(static_data) > 0: data["staticData"] = static_data workflow = await client.create_workflow(data) result = { "success": True, "message": f"Workflow '{name}' created successfully", "workflow": { "id": workflow.get("id"), "name": workflow.get("name"), "active": workflow.get("active", False), }, } return json.dumps(result, indent=2) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def update_workflow( client: N8nClient, workflow_id: str, name: str | None = None, nodes: list[dict[str, Any]] | None = None, connections: dict[str, Any] | None = None, settings: dict[str, Any] | None = None, ) -> str: """Update an existing workflow""" try: # Get current workflow first current = await client.get_workflow(workflow_id) # Build update data data = { "name": name or current.get("name"), "nodes": nodes or current.get("nodes", []), "connections": connections or current.get("connections", {}), } if settings: data["settings"] = settings workflow = await client.update_workflow(workflow_id, data) result = { "success": True, "message": f"Workflow '{workflow.get('name')}' updated successfully", "workflow": { "id": workflow.get("id"), "name": workflow.get("name"), "active": workflow.get("active"), }, } return json.dumps(result, indent=2) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def delete_workflow(client: N8nClient, workflow_id: str) -> str: """Delete a workflow""" try: await client.delete_workflow(workflow_id) result = {"success": True, "message": f"Workflow {workflow_id} deleted successfully"} return json.dumps(result, indent=2) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def activate_workflow(client: N8nClient, workflow_id: str) -> str: """Activate a workflow""" try: workflow = await client.activate_workflow(workflow_id) result = { "success": True, "message": f"Workflow '{workflow.get('name')}' activated successfully", "workflow": { "id": workflow.get("id"), "name": workflow.get("name"), "active": workflow.get("active"), }, } return json.dumps(result, indent=2) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def deactivate_workflow(client: N8nClient, workflow_id: str) -> str: """Deactivate a workflow""" try: workflow = await client.deactivate_workflow(workflow_id) result = { "success": True, "message": f"Workflow '{workflow.get('name')}' deactivated successfully", "workflow": { "id": workflow.get("id"), "name": workflow.get("name"), "active": workflow.get("active"), }, } return json.dumps(result, indent=2) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def execute_workflow(client: N8nClient, workflow_id: str) -> str: """Execute a workflow manually""" try: execution = await client.execute_workflow(workflow_id) result = { "success": True, "message": "Workflow execution started", "execution": { "id": execution.get("id"), "workflow_id": workflow_id, "status": execution.get("status", "running"), }, } return json.dumps(result, indent=2) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def execute_workflow_with_data( client: N8nClient, workflow_id: str, data: dict[str, Any] ) -> str: """Execute workflow with custom input data""" try: execution = await client.execute_workflow(workflow_id, data=data) result = { "success": True, "message": "Workflow execution started with custom data", "execution": { "id": execution.get("id"), "workflow_id": workflow_id, "status": execution.get("status", "running"), "input_data": data, }, } return json.dumps(result, indent=2) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def duplicate_workflow(client: N8nClient, workflow_id: str, new_name: str) -> str: """Duplicate a workflow""" try: # Get the original workflow original = await client.get_workflow(workflow_id) # Create new workflow with same structure data = { "name": new_name, "nodes": original.get("nodes", []), "connections": original.get("connections", {}), "settings": original.get("settings", {}), } new_workflow = await client.create_workflow(data) result = { "success": True, "message": f"Workflow duplicated as '{new_name}'", "original": {"id": original.get("id"), "name": original.get("name")}, "duplicate": {"id": new_workflow.get("id"), "name": new_workflow.get("name")}, } return json.dumps(result, indent=2) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def export_workflow(client: N8nClient, workflow_id: str) -> str: """Export workflow as JSON""" try: workflow = await client.get_workflow(workflow_id) # Return full workflow JSON for export result = {"success": True, "workflow_json": workflow} return json.dumps(result, indent=2) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def import_workflow( client: N8nClient, workflow_json: dict[str, Any], name_override: str | None = None ) -> str: """Import workflow from JSON""" try: # Use name override if provided if name_override: workflow_json["name"] = name_override # Ensure required fields if "nodes" not in workflow_json: workflow_json["nodes"] = [] if "connections" not in workflow_json: workflow_json["connections"] = {} workflow = await client.create_workflow(workflow_json) result = { "success": True, "message": f"Workflow '{workflow.get('name')}' imported successfully", "workflow": { "id": workflow.get("id"), "name": workflow.get("name"), "active": workflow.get("active", False), }, } return json.dumps(result, indent=2) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def get_workflow_tags(client: N8nClient, workflow_id: str) -> str: """Get tags assigned to a workflow""" try: tags = await client.get_workflow_tags(workflow_id) result = {"success": True, "workflow_id": workflow_id, "tags": tags} return json.dumps(result, indent=2) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2) async def set_workflow_tags( client: N8nClient, workflow_id: str, tag_ids: list[str] | None = None ) -> str: """Set tags for a workflow""" try: # Use empty list if tag_ids is None tags_to_set = tag_ids if tag_ids is not None else [] # Get current workflow current = await client.get_workflow(workflow_id) # Update with new tags data = { "name": current.get("name"), "nodes": current.get("nodes", []), "connections": current.get("connections", {}), "tags": [{"id": tid} for tid in tags_to_set], } workflow = await client.update_workflow(workflow_id, data) result = { "success": True, "message": f"Tags updated for workflow {workflow_id}", "workflow_id": workflow_id, "tags": [t.get("name") for t in workflow.get("tags", [])], } return json.dumps(result, indent=2) except Exception as e: return json.dumps({"success": False, "error": str(e)}, indent=2)