swarm-zap/skills/n8n-webhook/scripts/validate-workflow.py

#!/usr/bin/env python3
import json
import sys
from pathlib import Path

REQUIRED_NODE_NAMES = {
    'Webhook',
    'route-action',
    'route-dispatch',
    'Send Telegram Notification',
    'Send Discord Notification',
    'Respond to Webhook',
}

EXPECTED_TYPES = {
    'Webhook': 'n8n-nodes-base.webhook',
    'route-action': 'n8n-nodes-base.code',
    'route-dispatch': 'n8n-nodes-base.switch',
    'Send Telegram Notification': 'n8n-nodes-base.telegram',
    'Send Discord Notification': 'n8n-nodes-base.httpRequest',
    'Respond to Webhook': 'n8n-nodes-base.respondToWebhook',
}

SAMPLE_FILES = [
    'test-append-log.json',
    'test-notify.json',
    'test-send-notification-draft.json',
    'test-send-email-draft.json',
    'test-list-email-drafts.json',
    'test-delete-email-draft.json',
    'test-send-gmail-draft.json',
    'test-send-approved-email.json',
    'test-create-calendar-event.json',
    'test-list-upcoming-events.json',
    'test-update-calendar-event.json',
    'test-delete-calendar-event.json',
    'test-verify-email-draft-cycle.json',
    'test-verify-calendar-event-cycle.json',
    'test-fetch-and-normalize-url.json',
    'test-approval-queue-list.json',
    'test-inbound-event-filter.json',
]

ROUTER_SNIPPETS = [
    'append_log',
    'get_logs',
    'notify',
    'send_notification_draft',
    'send_email_draft',
    'list_email_drafts',
    'delete_email_draft',
    'send_gmail_draft',
    'send_approved_email',
    'create_calendar_event',
    'list_upcoming_events',
    'update_calendar_event',
    'delete_calendar_event',
    'approval_queue_add',
    'approval_queue_list',
    'approval_queue_resolve',
    'approval_history_attach_execution',
    'fetch_and_normalize_url',
    'inbound_event_filter',
    'unknown_action',
    'invalid_request',
    '$getWorkflowStaticData',
    'approvalQueue',
    'approvalHistory',
    'email_draft_send',
    'email_draft_delete',
    'email_list_drafts',
    'calendar_list_events',
    'calendar_event_update',
    'calendar_event_delete',
    'makeApprovalPolicy',
    'pending_compact',
    'history_compact',
    'summary_line',
    'result_refs',
    'default_mode',
    'inboundEvents',
    'eventDedup',
    'notify_text',
]


def fail(msg: str):
    print(f'ERROR: {msg}', file=sys.stderr)
    raise SystemExit(1)


def load_json(path: Path):
    try:
        return json.loads(path.read_text())
    except Exception as e:
        fail(f'failed to parse {path}: {e}')


def main():
    path = Path(sys.argv[1]) if len(sys.argv) > 1 else Path('assets/openclaw-action.workflow.json')
    data = load_json(path)
    nodes = data.get('nodes')
    connections = data.get('connections')
    if not isinstance(nodes, list):
        fail('workflow is missing a nodes array')
    if not isinstance(connections, dict):
        fail('workflow is missing a connections object')

    by_name = {}
    for node in nodes:
        if not isinstance(node, dict):
            fail('every node entry must be an object')
        name = node.get('name')
        if not name:
            fail('every node must have a name')
        by_name[name] = node

    missing = sorted(REQUIRED_NODE_NAMES - set(by_name))
    if missing:
        fail(f'missing required nodes: {", ".join(missing)}')

    for name, node_type in EXPECTED_TYPES.items():
        actual = by_name[name].get('type')
        if actual != node_type:
            fail(f'node {name!r} should have type {node_type!r}, got {actual!r}')

    webhook = by_name['Webhook'].get('parameters', {})
    if webhook.get('path') != 'openclaw-action':
        fail('Webhook.path must be openclaw-action')
    if webhook.get('httpMethod') != 'POST':
        fail('Webhook.httpMethod must be POST')
    if webhook.get('responseMode') != 'responseNode':
        fail('Webhook.responseMode must be responseNode')

    router = by_name['route-action'].get('parameters', {})
    js_code = router.get('jsCode', '')
    for snippet in ROUTER_SNIPPETS:
        if snippet not in js_code:
            fail(f'route-action jsCode missing expected snippet: {snippet!r}')

    switch = by_name['route-dispatch'].get('parameters', {})
    values = switch.get('rules', {}).get('values', [])
    names = {v.get('outputKey') for v in values if isinstance(v, dict)}
    if 'notify' not in names:
        fail('route-dispatch must route notify')

    telegram = by_name['Send Telegram Notification']
    if telegram.get('credentials', {}).get('telegramApi', {}).get('name') != 'Telegram Bot (OpenClaw)':
        fail('Send Telegram Notification must use Telegram Bot (OpenClaw) credential')

    discord = by_name['Send Discord Notification']
    if discord.get('credentials', {}).get('httpHeaderAuth', {}).get('name') != 'Discord Bot Auth':
        fail('Send Discord Notification must use Discord Bot Auth credential')

    responder = by_name['Respond to Webhook'].get('parameters', {})
    if responder.get('respondWith') != 'json':
        fail('Respond to Webhook must respondWith json')

    for sample_name in SAMPLE_FILES:
        sample = path.parent / sample_name
        sample_data = load_json(sample)
        if not isinstance(sample_data, dict) or 'action' not in sample_data or 'args' not in sample_data:
            fail(f'sample payload missing action/args: {sample}')

    print('OK: workflow asset structure looks consistent')
    print(f'- workflow: {path}')
    print(f'- nodes: {len(nodes)}')
    print('- routes: notify via Telegram + Discord; queue/log/fetch/filter handled in route-action code')
    print('- samples: ' + ', '.join(SAMPLE_FILES))


if __name__ == '__main__':
    main()