N8N Workflow Deployment

Deploy workflows to N8N using the REST API.

Environment Variables

N8N_HOST="https://your-n8n-instance.com" # Or http://localhost:5678 N8N_API_KEY="your-api-key" # From N8N Settings > API

Quick Deploy

import os import json from urllib.request import Request, urlopen from urllib.error import HTTPError

def deploy_n8n_workflow(workflow: dict, activate: bool = False) -> dict: """ Deploy a workflow to N8N.

Args:
    workflow: Complete workflow JSON with nodes, connections, settings
    activate: Whether to activate the workflow after creation

Returns:
    dict with id, name, and URL of created workflow
"""
host = os.environ.get('N8N_HOST', 'http://localhost:5678')
api_key = os.environ.get('N8N_API_KEY')

if not api_key:
    raise ValueError("N8N_API_KEY environment variable not set")

# Ensure workflow has required fields
if 'name' not in workflow:
    workflow['name'] = 'Untitled Workflow'
if 'nodes' not in workflow:
    workflow['nodes'] = []
if 'connections' not in workflow:
    workflow['connections'] = {}
if 'settings' not in workflow:
    workflow['settings'] = {}

# Remove 'active' field if present - it's read-only on creation
workflow.pop('active', None)

url = f"{host.rstrip('/')}/api/v1/workflows"

request = Request(
    url,
    data=json.dumps(workflow).encode('utf-8'),
    headers={
        'Content-Type': 'application/json',
        'X-N8N-API-KEY': api_key
    },
    method='POST'
)

with urlopen(request, timeout=30) as response:
    result = json.loads(response.read().decode('utf-8'))

workflow_id = result.get('id')
workflow_url = f"{host}/workflow/{workflow_id}"

print(f"✅ Workflow deployed to N8N!")
print(f"   ID: {workflow_id}")
print(f"   Name: {result.get('name')}")
print(f"   URL: {workflow_url}")

# Activate if requested (requires separate PATCH call)
is_active = False
if activate:
    is_active = set_workflow_active(workflow_id, True).get('active', False)
    print(f"   Active: {is_active}")
else:
    print(f"   Active: False (not activated)")

return {
    'id': workflow_id,
    'name': result.get('name'),
    'active': is_active,
    'url': workflow_url
}

Activate/Deactivate Workflow

def set_workflow_active(workflow_id: str, active: bool = True) -> dict: """Activate or deactivate an existing workflow.""" host = os.environ.get('N8N_HOST', 'http://localhost:5678') api_key = os.environ.get('N8N_API_KEY')

url = f"{host.rstrip('/')}/api/v1/workflows/{workflow_id}"

request = Request(
    url,
    data=json.dumps({'active': active}).encode('utf-8'),
    headers={
        'Content-Type': 'application/json',
        'X-N8N-API-KEY': api_key
    },
    method='PATCH'
)

with urlopen(request, timeout=30) as response:
    result = json.loads(response.read().decode('utf-8'))

status = "activated" if active else "deactivated"
print(f"✅ Workflow {workflow_id} {status}")
return result

List Workflows

def list_n8n_workflows(active_only: bool = False) -> list: """List all workflows in N8N instance.""" host = os.environ.get('N8N_HOST', 'http://localhost:5678') api_key = os.environ.get('N8N_API_KEY')

url = f"{host.rstrip('/')}/api/v1/workflows"
if active_only:
    url += "?active=true"

request = Request(
    url,
    headers={
        'Content-Type': 'application/json',
        'X-N8N-API-KEY': api_key
    }
)

with urlopen(request, timeout=30) as response:
    result = json.loads(response.read().decode('utf-8'))

workflows = result.get('data', [])
print(f"Found {len(workflows)} workflows")
for wf in workflows:
    status = "🟢" if wf.get('active') else "⚪"
    print(f"  {status} [{wf['id']}] {wf['name']}")

return workflows

Example: Anomaly Detection Workflow

Build an anomaly detection workflow

anomaly_workflow = { "name": "Daily Anomaly Detection", "nodes": [ { "id": "schedule", "name": "Daily 9am", "type": "n8n-nodes-base.scheduleTrigger", "typeVersion": 1, "position": [250, 300], "parameters": { "rule": { "interval": [{"field": "hours", "hoursInterval": 24}] } } }, { "id": "bigquery", "name": "Query Metrics", "type": "n8n-nodes-base.googleBigQuery", "typeVersion": 2, "position": [450, 300], "parameters": { "operation": "executeQuery", "projectId": "={{ $env.BIGQUERY_PROJECT_ID }}", "sqlQuery": "SELECT date, registration_starts, paid_conversions FROM metrics WHERE date >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)" } }, { "id": "code", "name": "Detect Anomalies", "type": "n8n-nodes-base.code", "typeVersion": 2, "position": [650, 300], "parameters": { "jsCode": ''' const data = $input.all(); const values = data.map(d => d.json.registration_starts); const mean = values.reduce((a,b) => a+b, 0) / values.length; const std = Math.sqrt(values.map(x => Math.pow(x - mean, 2)).reduce((a,b) => a+b, 0) / values.length);

const anomalies = data.filter(d => { const zscore = Math.abs((d.json.registration_starts - mean) / std); return zscore > 2; });

return anomalies.length > 0 ? anomalies : []; ''' } }, { "id": "if", "name": "Has Anomaly?", "type": "n8n-nodes-base.if", "typeVersion": 1, "position": [850, 300], "parameters": { "conditions": { "number": [{ "value1": "={{ $json.length }}", "operation": "larger", "value2": 0 }] } } }, { "id": "slack", "name": "Send Alert", "type": "n8n-nodes-base.slack", "typeVersion": 2, "position": [1050, 250], "parameters": { "channel": "#alerts", "text": "🔴 Anomaly detected in registration_starts" } } ], "connections": { "Daily 9am": {"main": [[{"node": "Query Metrics", "type": "main", "index": 0}]]}, "Query Metrics": {"main": [[{"node": "Detect Anomalies", "type": "main", "index": 0}]]}, "Detect Anomalies": {"main": [[{"node": "Has Anomaly?", "type": "main", "index": 0}]]}, "Has Anomaly?": {"main": [[{"node": "Send Alert", "type": "main", "index": 0}], []]} }, "settings": { "executionOrder": "v1" } }

Deploy it

result = deploy_n8n_workflow(anomaly_workflow, activate=False) print(f"Workflow ready at: {result['url']}")

cURL Examples

Create Workflow

curl -X POST "$N8N_HOST/api/v1/workflows"
-H "Content-Type: application/json"
-H "X-N8N-API-KEY: $N8N_API_KEY"
-d @workflow.json

List Workflows

curl -X GET "$N8N_HOST/api/v1/workflows"
-H "X-N8N-API-KEY: $N8N_API_KEY"

Activate Workflow

curl -X PATCH "$N8N_HOST/api/v1/workflows/WORKFLOW_ID"
-H "Content-Type: application/json"
-H "X-N8N-API-KEY: $N8N_API_KEY"
-d '{"active": true}'

Error Handling

from urllib.error import HTTPError

try: result = deploy_n8n_workflow(workflow) except HTTPError as e: error_body = e.read().decode('utf-8') print(f"N8N API error: {e.code}") print(f" {error_body}") except ValueError as e: print(f"Configuration error: {e}")

Getting Your N8N API Key

• Open your N8N instance

• Go to Settings (gear icon)

• Click API in the left menu

• Click Create API Key

• Copy the key and set it as N8N_API_KEY