N8n Development

n8n Development Skill

This skill provides comprehensive guidance for developing, testing, and maintaining n8n webhook workflows integrated with Python services. It includes patterns for building composable workflows, template development, test-driven development practices, and complete development workflows.

When to Use This Skill

Apply this skill when:

• Building new n8n webhook workflows
• Integrating n8n workflows with Python services
• Writing tests for workflow integrations
• Creating reusable workflow templates
• Debugging workflow execution issues
• Setting up development environments for n8n projects
• Following best practices for async patterns and error handling

Core Development Principles

Test-Driven Development

1. Start with a failing test to define expected behavior 2. Implement minimal code to make the test pass 3. Refactor while keeping tests green 4. Run tests with uv run pytest

Workflow Development Process

1. Design the webhook contract (request/response schema) 2. Implement in n8n with error handling 3. Export JSON to nathan/workflows/[category]/ 4. Add to registry (workflow_registry.yaml) 5. Write integration tests in tests/workflows/[category]/ 6. Document with usage examples

Template Development

1. Define variables using ${VAR_NAME} syntax 2. Add validation via schema in template frontmatter 3. Test rendering using CLI 4. Document with usage examples in comments

Essential Commands

# Core development
uv sync                                    # Install dependencies
uv run pytest                              # Run tests
uv run pytest --cov=nathan --cov-report=term-missing  # With coverage
uvx ruff check .                           # Lint
uvx ruff format .                          # Format

# n8n development
docker compose -f docker-compose.n8n.yml up -d  # Run n8n locally
uv run python -m nathan.scripts.n8n_workflow_registry --help  # Registry CLI
uv run python -m nathan.templating --help  # Template CLI

Testing Guidelines

Test structure should mirror source in tests/ directory with descriptive naming (test_trigger_workflow_with_valid_parameters). Use pytest fixtures for shared setup and mock external dependencies. Target 80%+ test coverage.

Validation and Schemas

Use Pydantic models for API requests/responses and SQLModel for database models. Implement input validation at boundaries (API endpoints, CLI arguments) with clear, actionable error messages.

Async Patterns

Use httpx for all HTTP calls with proper timeout handling. Always use try/except for HTTP calls and async with context managers for HTTP clients. Use asyncio.gather() for parallel operations.

Reference Documents

Load these reference files when needed for specific tasks:

• Common Patterns: Load references/common_patterns.md for workflow registry, template rendering, and error handling code examples
• Code Review: Load references/review_checklist.md before submitting code for review
• CLI Commands: Load references/cli_commands.md for n8n command-line operations (export, import, execute workflows)
• REST API: Load references/rest_api.md for n8n REST API endpoints and Python client examples
• Workflow JSON: Load references/workflow_json_structure.md for understanding and building n8n workflow JSON files

Commit Guidelines

Follow conventional commits format:

• feat: for new features
• fix: for bug fixes
• docs: for documentation changes
• test: for test additions/changes
• refactor: for code refactoring

Make atomic commits (one logical change per commit), test before committing, and include documentation updates with code changes.

n8n CLI Commands Reference

This document provides a comprehensive reference for n8n command-line interface operations, including official n8n CLI commands and third-party tools.

Official n8n CLI Commands

The n8n CLI is built into the n8n application and provides commands for workflow management, credential handling, and instance administration.

Running CLI Commands

Local Installation:

n8n <command> [options]

Docker:

docker exec -u node -it <n8n-container-name> n8n <command> [options]

Docker Compose:

docker compose exec n8n n8n <command> [options]

Workflow Management

Export Workflows

Export workflows to JSON files for backup, version control, or migration.

# Export single workflow by ID
n8n export:workflow --id=<workflow_id> --output=workflow.json

# Export all workflows to separate files
n8n export:workflow --backup --output=backups/latest/

# Export all workflows to a single file
n8n export:workflow --all --output=all_workflows.json

# Export workflow without credential data
n8n export:workflow --id=<workflow_id> --output=workflow.json --decrypted

Options:

Import Workflows

Import workflows from JSON files.

# Import single workflow
n8n import:workflow --input=workflow.json

# Import multiple workflows from separate files
n8n import:workflow --separate --input=backups/latest/

# Import with credential updates
n8n import:workflow --input=workflow.json --userId=<user_id>

Options:

Execute Workflows

Execute a workflow directly from the command line.

# Execute workflow by ID
n8n execute --id=<workflow_id>

# Execute with input data
n8n execute --id=<workflow_id> --rawInput='{"key": "value"}'

# Execute from file
n8n execute --file=workflow.json

Options:

Update Workflow Status

Activate or deactivate workflows.

# Deactivate a specific workflow
n8n update:workflow --id=<workflow_id> --active=false

# Activate a specific workflow
n8n update:workflow --id=<workflow_id> --active=true

# Activate all workflows
n8n update:workflow --all --active=true

# Deactivate all workflows
n8n update:workflow --all --active=false

Credential Management

Export Credentials

# Export all credentials
n8n export:credentials --backup --output=credentials/

# Export specific credential by ID
n8n export:credentials --id=<credential_id> --output=credential.json

# Export decrypted credentials (requires ENCRYPTION_KEY)
n8n export:credentials --backup --output=credentials/ --decrypted

Import Credentials

# Import credentials
n8n import:credentials --input=credentials/

# Import with user assignment
n8n import:credentials --input=credentials/ --userId=<user_id>

User Management

# Reset user password
n8n user-management:reset

# List users (when using user management)
n8n user-management:list

Database Operations

# Run database migrations
n8n db:revert

# Check database status
n8n db:check

Instance Information

# Get n8n version
n8n --version

# Get help
n8n --help

# Get help for specific command
n8n export:workflow --help

Third-Party CLI Tool: n8n-cli

The n8n-cli is a community-developed tool for managing n8n workflows via the API.

Installation

# Using curl (Linux/macOS)
curl -sSLf https://raw.github.com/edenreich/n8n-cli/main/install.sh | sh

# Using Homebrew
brew install edenreich/tap/n8n-cli

# Using Go
go install github.com/edenreich/n8n-cli@latest

Configuration

# Set n8n instance URL
export N8N_BASE_URL=http://localhost:5678

# Set API key
export N8N_API_KEY=your-api-key

Or create a config file at ~/.n8n-cli.yaml:

base_url: http://localhost:5678
api_key: your-api-key

Workflow Commands

# List all workflows
n8n workflows list

# Get workflow details
n8n workflows get <workflow_id>

# Sync workflows to local directory
n8n workflows sync --directory workflows/

# Refresh workflows from n8n to local directory
n8n workflows refresh --directory workflows/

# Push local workflows to n8n
n8n workflows push --directory workflows/

# Activate workflow
n8n workflows activate <workflow_id>

# Deactivate workflow
n8n workflows deactivate <workflow_id>

# Delete workflow
n8n workflows delete <workflow_id>

Execution Commands

# List executions
n8n executions list

# Get execution details
n8n executions get <execution_id>

# Delete execution
n8n executions delete <execution_id>

Common Patterns

Backup Workflow

Create a backup script for n8n workflows:

#!/bin/bash
# backup-n8n.sh

BACKUP_DIR="backups/$(date +%Y-%m-%d)"
mkdir -p "$BACKUP_DIR"

# Export all workflows
n8n export:workflow --backup --output="$BACKUP_DIR/workflows/"

# Export all credentials (encrypted)
n8n export:credentials --backup --output="$BACKUP_DIR/credentials/"

echo "Backup completed to $BACKUP_DIR"

CI/CD Integration

Deploy workflows from version control:

#!/bin/bash
# deploy-workflows.sh

# Deactivate all workflows before import
n8n update:workflow --all --active=false

# Import workflows from repository
n8n import:workflow --separate --input=workflows/

# Reactivate required workflows
n8n update:workflow --id=<workflow_id> --active=true

echo "Deployment completed"

Development Workflow

Sync workflows between local development and n8n:

# Export workflow after editing in n8n UI
n8n export:workflow --id=<workflow_id> --output=nathan/workflows/jira/my-workflow.json

# Import workflow after editing JSON locally
n8n import:workflow --input=nathan/workflows/jira/my-workflow.json

Environment Variables

Troubleshooting

Common Issues

Permission denied when running CLI in Docker:

# Use the node user
docker exec -u node -it n8n n8n <command>

Workflow import fails with credential errors:

Exported workflows reference credentials by ID. After import, you may need to:

1. Re-create credentials in the target n8n instance 2. Update workflow nodes to use the new credential IDs 3. Or use the --decrypted flag and ensure the same ENCRYPTION_KEY

Workflow not found:

Ensure you're using the correct workflow ID (numeric ID, not the name):

# List workflows to get IDs
n8n export:workflow --all --output=/dev/stdout | jq '.[] | {id, name}'

Related Documentation

• Official n8n CLI Documentation
• n8n REST API Reference
• Workflow JSON Structure

Common n8n Development Patterns

Workflow Registry Pattern

Basic Usage

from nathan.core.workflow_registry import N8NWorkflowRegistry, trigger_n8n_workflow

# List available workflows
registry = N8NWorkflowRegistry()
workflows = registry.list_all()

# Trigger a workflow
result = await trigger_n8n_workflow(
    workflow_url="http://localhost:5678/webhook/get-jira-ticket",
    parameters={"ticket_id": "PROJ-123"},
    shared_secret="your-secret",
)

if result.success:
    print(result.data)
else:
    print(f"Error: {result.error}")

Registry Configuration

# workflow_registry.yaml
workflows:
  - id: get-jira-ticket
    name: Get Jira Ticket
    webhook_path: /webhook/get-jira-ticket
    description: Fetch ticket details from Jira
    parameters:
      - name: ticket_id
        type: string
        required: true
        description: Jira ticket ID (e.g., PROJ-123)

Template Rendering Pattern

Basic Template Rendering

from nathan.templating.api import render_template

rendered = render_template(
    template_path="templates/flows/jira-workflow.yaml",
    variables={"project_key": "PROJ", "issue_type": "Task"},
)

with open("output.json", "w") as f:
    f.write(rendered)

Template with Validation

from nathan.templating.api import render_template, validate_template

# Validate before rendering
is_valid = validate_template(
    template_path="templates/flows/jira-workflow.yaml",
    variables={"project_key": "PROJ", "issue_type": "Task"},
)

if is_valid:
    rendered = render_template(
        template_path="templates/flows/jira-workflow.yaml",
        variables={"project_key": "PROJ", "issue_type": "Task"},
    )

Error Handling Patterns

Comprehensive Error Handling

from nathan.templating.exceptions import ValidationError, TemplateError
from nathan.core.exceptions import WorkflowExecutionError
import httpx

async def execute_workflow_safely(url: str, params: dict, secret: str):
    """Execute workflow with proper error handling."""
    try:
        result = await trigger_n8n_workflow(url, params, secret)
        if not result.success:
            raise WorkflowExecutionError(
                f"Workflow failed: {result.error}",
                status_code=result.status_code
            )
        return result.data
    except httpx.TimeoutException:
        logger.error(f"Workflow timed out: {url}")
        raise WorkflowExecutionError("Workflow execution timed out")
    except httpx.RequestError as e:
        logger.error(f"HTTP request failed: {e}")
        raise WorkflowExecutionError(f"HTTP request failed: {str(e)}")
    except ValidationError as e:
        logger.error(f"Validation failed: {e}")
        raise

Retry Pattern with Exponential Backoff

import asyncio
from typing import Optional

async def execute_with_retry(
    url: str,
    params: dict,
    secret: str,
    max_retries: int = 3,
    base_delay: float = 1.0,
) -> Optional[dict]:
    """Execute workflow with exponential backoff retry."""
    for attempt in range(max_retries):
        try:
            result = await trigger_n8n_workflow(url, params, secret)
            if result.success:
                return result.data

            # Don't retry on client errors (4xx)
            if 400 <= result.status_code < 500:
                raise WorkflowExecutionError(f"Client error: {result.error}")

        except httpx.TimeoutException:
            if attempt == max_retries - 1:
                raise

        # Exponential backoff
        delay = base_delay * (2 ** attempt)
        await asyncio.sleep(delay)

    return None

Testing Patterns

Mocking HTTP Calls

import pytest
from unittest.mock import AsyncMock, patch

@pytest.mark.asyncio
async def test_trigger_workflow_success():
    """Test successful workflow trigger."""
    mock_response = AsyncMock()
    mock_response.status_code = 200
    mock_response.json.return_value = {"success": True, "data": {"id": "123"}}

    with patch("httpx.AsyncClient.post", return_value=mock_response):
        result = await trigger_n8n_workflow(
            "http://localhost:5678/webhook/test",
            {"param": "value"},
            "secret"
        )

        assert result.success
        assert result.data["id"] == "123"

Testing with Fixtures

import pytest
from nathan.core.workflow_registry import N8NWorkflowRegistry

@pytest.fixture
def registry():
    """Provide a workflow registry instance."""
    return N8NWorkflowRegistry(config_path="tests/fixtures/test_registry.yaml")

@pytest.fixture
def mock_n8n_server():
    """Mock n8n server responses."""
    with patch("httpx.AsyncClient") as mock_client:
        mock_instance = AsyncMock()
        mock_client.return_value = mock_instance
        yield mock_instance

@pytest.mark.asyncio
async def test_list_workflows(registry):
    """Test listing available workflows."""
    workflows = registry.list_all()
    assert len(workflows) > 0
    assert all(w.id for w in workflows)

Async Patterns

Parallel Workflow Execution

import asyncio
from typing import List, Dict

async def execute_parallel_workflows(
    workflows: List[Dict[str, any]],
    shared_secret: str
) -> List[dict]:
    """Execute multiple workflows in parallel."""
    tasks = [
        trigger_n8n_workflow(
            workflow["url"],
            workflow["params"],
            shared_secret
        )
        for workflow in workflows
    ]

    results = await asyncio.gather(*tasks, return_exceptions=True)

    # Process results
    successful_results = []
    for result in results:
        if isinstance(result, Exception):
            logger.error(f"Workflow failed: {result}")
        elif result.success:
            successful_results.append(result.data)
        else:
            logger.error(f"Workflow error: {result.error}")

    return successful_results

Context Manager for HTTP Client

import httpx
from contextlib import asynccontextmanager

@asynccontextmanager
async def n8n_client(base_url: str, timeout: float = 30.0):
    """Context manager for n8n HTTP client."""
    async with httpx.AsyncClient(
        base_url=base_url,
        timeout=httpx.Timeout(timeout),
        headers={"Content-Type": "application/json"}
    ) as client:
        yield client

# Usage
async def fetch_workflow_status(workflow_id: str):
    async with n8n_client("http://localhost:5678") as client:
        response = await client.get(f"/workflow/{workflow_id}")
        return response.json()

Pydantic Models

Request/Response Models

from pydantic import BaseModel, Field, validator
from typing import Optional, Dict, Any
from datetime import datetime

class WorkflowRequest(BaseModel):
    """Request model for workflow execution."""
    workflow_id: str = Field(..., description="Workflow identifier")
    parameters: Dict[str, Any] = Field(default_factory=dict)
    timeout: Optional[int] = Field(30, ge=1, le=300)

    @validator("workflow_id")
    def validate_workflow_id(cls, v):
        if not v or not v.strip():
            raise ValueError("workflow_id cannot be empty")
        return v.strip()

class WorkflowResponse(BaseModel):
    """Response model for workflow execution."""
    success: bool
    data: Optional[Dict[str, Any]] = None
    error: Optional[str] = None
    status_code: int = 200
    timestamp: datetime = Field(default_factory=datetime.utcnow)

    class Config:
        json_encoders = {
            datetime: lambda v: v.isoformat()
        }

Template Models

from pydantic import BaseModel, Field
from typing import List, Dict, Any

class TemplateVariable(BaseModel):
    """Template variable definition."""
    name: str
    type: str = Field(..., regex="^(string|number|boolean|array|object)$")
    required: bool = True
    default: Optional[Any] = None
    description: Optional[str] = None

class TemplateSchema(BaseModel):
    """Template schema definition."""
    variables: List[TemplateVariable]
    version: str = "1.0.0"
    description: Optional[str] = None

    def validate_variables(self, provided: Dict[str, Any]) -> bool:
        """Validate provided variables against schema."""
        for var in self.variables:
            if var.required and var.name not in provided:
                raise ValidationError(f"Required variable '{var.name}' not provided")

            if var.name in provided:
                # Type validation logic here
                pass

        return True

CLI Integration

Click Command Pattern

import click
from pathlib import Path

@click.command()
@click.argument("template_path", type=click.Path(exists=True))
@click.option("--variables", "-v", multiple=True, help="Variables as key=value")
@click.option("--output", "-o", type=click.Path(), help="Output file path")
def render(template_path: str, variables: tuple, output: str):
    """Render an n8n workflow template."""
    # Parse variables
    vars_dict = {}
    for var in variables:
        key, value = var.split("=", 1)
        vars_dict[key] = value

    # Render template
    try:
        rendered = render_template(template_path, vars_dict)

        if output:
            Path(output).write_text(rendered)
            click.echo(f"Template rendered to {output}")
        else:
            click.echo(rendered)
    except ValidationError as e:
        click.echo(f"Validation error: {e}", err=True)
        raise click.Exit(1)

Integration Testing

Testing n8n Workflows

import pytest
import docker
from pathlib import Path

@pytest.fixture(scope="session")
def n8n_container():
    """Start n8n container for integration tests."""
    client = docker.from_env()

    # Start n8n container
    container = client.containers.run(
        "n8nio/n8n",
        ports={"5678/tcp": 5678},
        environment={
            "N8N_BASIC_AUTH_ACTIVE": "false",
            "N8N_HOST": "localhost",
        },
        detach=True,
        remove=True,
    )

    # Wait for n8n to be ready
    import time
    time.sleep(10)

    yield container

    # Cleanup
    container.stop()

@pytest.mark.integration
async def test_workflow_execution(n8n_container):
    """Test actual workflow execution."""
    result = await trigger_n8n_workflow(
        "http://localhost:5678/webhook/test",
        {"message": "test"},
        "test-secret"
    )

    assert result.success
    assert result.data["message"] == "test"

n8n REST API Reference

This document provides a comprehensive reference for the n8n REST API, enabling programmatic management of workflows, executions, credentials, and other n8n resources.

API Overview

n8n provides a public REST API for performing GUI tasks programmatically. The API is available at:

https://<your-n8n-instance>/api/v1/

Authentication

All API requests require authentication via API key:

# Header-based authentication
curl -H "X-N8N-API-KEY: your-api-key" \
     https://your-n8n.example.com/api/v1/workflows

API Key Generation

1. Open n8n UI 2. Go to Settings > API 3. Click Create API Key 4. Copy and securely store the key

Workflows API

List Workflows

GET /api/v1/workflows

Query Parameters:

Example:

curl -H "X-N8N-API-KEY: your-api-key" \
     "https://your-n8n.example.com/api/v1/workflows?active=true&limit=10"

Response:

{
  "data": [
    {
      "id": "1",
      "name": "My Workflow",
      "active": true,
      "createdAt": "2024-01-01T00:00:00.000Z",
      "updatedAt": "2024-01-02T00:00:00.000Z",
      "tags": []
    }
  ],
  "nextCursor": "eyJpZCI6IjIifQ=="
}

Get Workflow

GET /api/v1/workflows/{id}

Example:

curl -H "X-N8N-API-KEY: your-api-key" \
     "https://your-n8n.example.com/api/v1/workflows/1"

Response:

{
  "id": "1",
  "name": "My Workflow",
  "active": true,
  "nodes": [...],
  "connections": {...},
  "settings": {...},
  "staticData": null,
  "createdAt": "2024-01-01T00:00:00.000Z",
  "updatedAt": "2024-01-02T00:00:00.000Z"
}

Create Workflow

POST /api/v1/workflows

Request Body:

{
  "name": "New Workflow",
  "nodes": [
    {
      "parameters": {},
      "id": "start",
      "name": "Start",
      "type": "n8n-nodes-base.start",
      "typeVersion": 1,
      "position": [250, 300]
    }
  ],
  "connections": {},
  "settings": {
    "executionOrder": "v1"
  }
}

Example:

curl -X POST \
     -H "X-N8N-API-KEY: your-api-key" \
     -H "Content-Type: application/json" \
     -d '{"name": "New Workflow", "nodes": [...], "connections": {}}' \
     "https://your-n8n.example.com/api/v1/workflows"

Update Workflow

PATCH /api/v1/workflows/{id}

Request Body:

{
  "name": "Updated Workflow Name",
  "active": false
}

Example:

curl -X PATCH \
     -H "X-N8N-API-KEY: your-api-key" \
     -H "Content-Type: application/json" \
     -d '{"name": "Updated Name"}' \
     "https://your-n8n.example.com/api/v1/workflows/1"

Delete Workflow

DELETE /api/v1/workflows/{id}

Example:

curl -X DELETE \
     -H "X-N8N-API-KEY: your-api-key" \
     "https://your-n8n.example.com/api/v1/workflows/1"

Activate Workflow

POST /api/v1/workflows/{id}/activate

Example:

curl -X POST \
     -H "X-N8N-API-KEY: your-api-key" \
     "https://your-n8n.example.com/api/v1/workflows/1/activate"

Deactivate Workflow

POST /api/v1/workflows/{id}/deactivate

Example:

curl -X POST \
     -H "X-N8N-API-KEY: your-api-key" \
     "https://your-n8n.example.com/api/v1/workflows/1/deactivate"

Executions API

List Executions

GET /api/v1/executions

Query Parameters:

Example:

curl -H "X-N8N-API-KEY: your-api-key" \
     "https://your-n8n.example.com/api/v1/executions?workflowId=1&status=error"

Get Execution

GET /api/v1/executions/{id}

Query Parameters:

Example:

curl -H "X-N8N-API-KEY: your-api-key" \
     "https://your-n8n.example.com/api/v1/executions/123?includeData=true"

Delete Execution

DELETE /api/v1/executions/{id}

Example:

curl -X DELETE \
     -H "X-N8N-API-KEY: your-api-key" \
     "https://your-n8n.example.com/api/v1/executions/123"

Credentials API

List Credentials

GET /api/v1/credentials

Example:

curl -H "X-N8N-API-KEY: your-api-key" \
     "https://your-n8n.example.com/api/v1/credentials"

Response:

{
  "data": [
    {
      "id": "1",
      "name": "Jira API",
      "type": "jiraApi",
      "createdAt": "2024-01-01T00:00:00.000Z",
      "updatedAt": "2024-01-02T00:00:00.000Z"
    }
  ]
}

Create Credential

POST /api/v1/credentials

Request Body:

{
  "name": "My Jira Credential",
  "type": "jiraApi",
  "data": {
    "email": "user@example.com",
    "apiToken": "your-token",
    "domain": "yourcompany.atlassian.net"
  }
}

Delete Credential

DELETE /api/v1/credentials/{id}

Tags API

List Tags

GET /api/v1/tags

Create Tag

POST /api/v1/tags

Request Body:

{
  "name": "production"
}

Webhooks

Trigger Webhook

Webhooks are not part of the REST API but are workflow-specific endpoints:

POST /webhook/{webhook-path}

Example with shared secret:

curl -X POST \
     -H "Content-Type: application/json" \
     -H "X-N8N-SECRET: your-shared-secret" \
     -d '{"ticket_id": "PROJ-123"}' \
     "https://your-n8n.example.com/webhook/get-jira-ticket"

Test Webhook

For development/testing, n8n provides test webhook endpoints:

POST /webhook-test/{webhook-path}

Test webhooks are only active when the workflow editor is open and listening.

Python Client Example

import httpx
from typing import Optional
from dataclasses import dataclass

@dataclass
class N8NClient:
    """Client for n8n REST API."""

    base_url: str
    api_key: str
    timeout: float = 30.0

    async def list_workflows(
        self,
        active: Optional[bool] = None,
        limit: int = 100
    ) -> dict:
        """List all workflows."""
        params = {"limit": limit}
        if active is not None:
            params["active"] = str(active).lower()

        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.get(
                f"{self.base_url}/api/v1/workflows",
                headers={"X-N8N-API-KEY": self.api_key},
                params=params
            )
            response.raise_for_status()
            return response.json()

    async def get_workflow(self, workflow_id: str) -> dict:
        """Get workflow by ID."""
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.get(
                f"{self.base_url}/api/v1/workflows/{workflow_id}",
                headers={"X-N8N-API-KEY": self.api_key}
            )
            response.raise_for_status()
            return response.json()

    async def create_workflow(self, workflow_data: dict) -> dict:
        """Create a new workflow."""
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.post(
                f"{self.base_url}/api/v1/workflows",
                headers={
                    "X-N8N-API-KEY": self.api_key,
                    "Content-Type": "application/json"
                },
                json=workflow_data
            )
            response.raise_for_status()
            return response.json()

    async def activate_workflow(self, workflow_id: str) -> dict:
        """Activate a workflow."""
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.post(
                f"{self.base_url}/api/v1/workflows/{workflow_id}/activate",
                headers={"X-N8N-API-KEY": self.api_key}
            )
            response.raise_for_status()
            return response.json()

    async def trigger_webhook(
        self,
        webhook_path: str,
        data: dict,
        shared_secret: Optional[str] = None
    ) -> dict:
        """Trigger a webhook workflow."""
        headers = {"Content-Type": "application/json"}
        if shared_secret:
            headers["X-N8N-SECRET"] = shared_secret

        async with httpx.AsyncClient(timeout=self.timeout) as client:
            response = await client.post(
                f"{self.base_url}/webhook/{webhook_path}",
                headers=headers,
                json=data
            )
            response.raise_for_status()
            return response.json()


# Usage example
async def main():
    client = N8NClient(
        base_url="http://localhost:5678",
        api_key="your-api-key"
    )

    # List active workflows
    workflows = await client.list_workflows(active=True)
    print(f"Found {len(workflows['data'])} active workflows")

    # Trigger webhook
    result = await client.trigger_webhook(
        webhook_path="get-jira-ticket",
        data={"ticket_id": "PROJ-123"},
        shared_secret="your-shared-secret"
    )
    print(f"Webhook result: {result}")

Error Handling

Common HTTP Status Codes

Error Response Format

{
  "code": 404,
  "message": "Workflow not found",
  "hint": "The workflow with ID '123' does not exist"
}

Rate Limiting

n8n does not have built-in rate limiting on the REST API, but consider implementing client-side rate limiting for production use:

import asyncio
from functools import wraps

def rate_limit(calls_per_second: float = 10):
    """Rate limit decorator for async functions."""
    min_interval = 1.0 / calls_per_second
    last_call = [0.0]

    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            elapsed = asyncio.get_event_loop().time() - last_call[0]
            if elapsed < min_interval:
                await asyncio.sleep(min_interval - elapsed)
            last_call[0] = asyncio.get_event_loop().time()
            return await func(*args, **kwargs)
        return wrapper
    return decorator

Related Documentation

• n8n CLI Commands
• Workflow JSON Structure
• Official n8n API Docs

Code Review Checklist

Pre-Submission Checklist

Testing

• [ ] All tests pass: uv run pytest
• [ ] Test coverage meets target (80%+): uv run pytest --cov=nathan --cov-report=term-missing
• [ ] New functionality has corresponding tests
• [ ] Integration tests for workflow changes
• [ ] Edge cases and error conditions tested
• [ ] Mocked external dependencies appropriately

Code Quality

• [ ] Linting passes: uvx ruff check .
• [ ] Code formatted: uvx ruff format .
• [ ] No commented-out code or debug statements
• [ ] Clear variable and function names
• [ ] DRY principle followed (no duplicated code)
• [ ] Single responsibility principle maintained

Documentation

• [ ] Docstrings for all new functions/classes
• [ ] Type hints for all parameters and returns
• [ ] Complex logic has inline comments
• [ ] README updated if adding new features
• [ ] API documentation updated if changing interfaces
• [ ] Usage examples provided for new functionality

Error Handling

• [ ] All HTTP calls wrapped in try/except
• [ ] Timeout handling implemented
• [ ] Clear, actionable error messages
• [ ] Proper logging at appropriate levels
• [ ] Graceful degradation where applicable
• [ ] No silent failures

Async Patterns

• [ ] Using httpx for HTTP calls (not requests)
• [ ] Async context managers used properly
• [ ] No blocking I/O in async functions
• [ ] Proper use of asyncio.gather() for parallel operations
• [ ] Timeouts configured for all external calls

n8n Specific

• [ ] Webhook contract documented
• [ ] Request/response schemas defined with Pydantic
• [ ] Workflow JSON exported to correct directory
• [ ] Registry entry added to workflow_registry.yaml
• [ ] Shared secret authentication implemented
• [ ] Consistent error response format

Security

• [ ] No hardcoded secrets or credentials
• [ ] Environment variables used for configuration
• [ ] Input validation at all boundaries
• [ ] SQL injection prevention (if applicable)
• [ ] XSS prevention (if applicable)
• [ ] Secrets not logged

Performance

• [ ] No unnecessary database queries
• [ ] Efficient data structures used
• [ ] Pagination implemented for large datasets
• [ ] Caching considered where appropriate
• [ ] Resource cleanup (close connections, files)

Commit Message Review

Format

• [ ] Follows conventional commits format
• [ ] Subject line under 50 characters
• [ ] Imperative mood in subject line
• [ ] Body explains why, not what
• [ ] References related issues/tickets

Conventional Commit Types

• feat: New feature
• fix: Bug fix
• docs: Documentation only changes
• style: Code style changes (formatting, etc.)
• refactor: Code change that neither fixes a bug nor adds a feature
• perf: Performance improvement
• test: Adding or updating tests
• chore: Changes to build process or auxiliary tools
• ci: CI/CD configuration changes

Pull Request Review

PR Description

• [ ] Clear title describing the change
• [ ] Summary of what changed and why
• [ ] Links to related issues/tickets
• [ ] Breaking changes noted
• [ ] Migration steps provided if needed

PR Content

• [ ] Single logical change per PR
• [ ] No unrelated changes mixed in
• [ ] Appropriate size (prefer smaller PRs)
• [ ] Base branch is correct
• [ ] No merge conflicts

Workflow Testing Checklist

Unit Tests

• [ ] Test each workflow parameter
• [ ] Test required vs optional parameters
• [ ] Test parameter type validation
• [ ] Test error responses
• [ ] Test timeout scenarios

Integration Tests

• [ ] Test against running n8n instance
• [ ] Test webhook authentication
• [ ] Test complete workflow execution
• [ ] Test error propagation
• [ ] Test concurrent executions

Template Tests

• [ ] Variable substitution works correctly
• [ ] Invalid variables caught
• [ ] Template syntax validation
• [ ] Output format validation
• [ ] Edge cases (empty values, special characters)

Common Issues to Check

Python Specific

• [ ] No mutable default arguments
• [ ] Context managers used for resources
• [ ] Proper exception hierarchy
• [ ] No bare except clauses
• [ ] f-strings used for formatting (Python 3.6+)

Async Specific

• [ ] No synchronous I/O in async functions
• [ ] Proper cancellation handling
• [ ] No fire-and-forget tasks
• [ ] Proper exception propagation in gathered tasks
• [ ] Resource cleanup in finally blocks

n8n Workflow Specific

• [ ] Idempotent operations where possible
• [ ] Proper retry logic for transient failures
• [ ] Rate limiting considered
• [ ] Webhook URL validation
• [ ] Response size limits handled

Post-Review Actions

• [ ] Address all review comments
• [ ] Re-run tests after changes
• [ ] Update documentation if changes made
• [ ] Squash commits if requested
• [ ] Verify CI/CD passes
• [ ] Request re-review if significant changes made

n8n Workflow JSON Structure Reference

This document provides a comprehensive reference for the n8n workflow JSON format, essential for programmatic workflow creation, export/import operations, and version control.

JSON Structure Overview

An n8n workflow JSON file contains four main components:

{
  "name": "Workflow Name",
  "nodes": [],
  "connections": {},
  "settings": {},
  "staticData": null,
  "tags": [],
  "meta": {}
}

Top-Level Properties

Nodes Array

Each node in the nodes array has this structure:

{
  "id": "unique-node-id",
  "name": "Node Display Name",
  "type": "n8n-nodes-base.httpRequest",
  "typeVersion": 4.2,
  "position": [250, 300],
  "parameters": {},
  "credentials": {},
  "disabled": false,
  "notesInFlow": false,
  "notes": ""
}

Node Properties

Common Node Types

Triggers:
- n8n-nodes-base.webhook
- n8n-nodes-base.scheduleTrigger
- n8n-nodes-base.manualTrigger

Core:
- n8n-nodes-base.httpRequest
- n8n-nodes-base.code
- n8n-nodes-base.set
- n8n-nodes-base.if
- n8n-nodes-base.switch
- n8n-nodes-base.merge
- n8n-nodes-base.splitInBatches
- n8n-nodes-base.respondToWebhook

Integrations:
- n8n-nodes-base.jira
- n8n-nodes-base.slack
- n8n-nodes-base.googleSheets
- n8n-nodes-base.airtable

Node Examples

Webhook Node

{
  "id": "webhook-1",
  "name": "Webhook",
  "type": "n8n-nodes-base.webhook",
  "typeVersion": 2,
  "position": [250, 300],
  "webhookId": "abc-123-def",
  "parameters": {
    "path": "get-jira-ticket",
    "httpMethod": "POST",
    "responseMode": "responseNode",
    "options": {
      "rawBody": false
    }
  }
}

HTTP Request Node

{
  "id": "http-request-1",
  "name": "HTTP Request",
  "type": "n8n-nodes-base.httpRequest",
  "typeVersion": 4.2,
  "position": [450, 300],
  "parameters": {
    "method": "GET",
    "url": "https://api.example.com/data",
    "authentication": "genericCredentialType",
    "genericAuthType": "httpHeaderAuth",
    "sendHeaders": true,
    "headerParameters": {
      "parameters": [
        {
          "name": "Accept",
          "value": "application/json"
        }
      ]
    },
    "options": {
      "timeout": 30000,
      "response": {
        "response": {
          "responseFormat": "json"
        }
      }
    }
  },
  "credentials": {
    "httpHeaderAuth": {
      "id": "1",
      "name": "My API Key"
    }
  }
}

Jira Node

{
  "id": "jira-1",
  "name": "Jira",
  "type": "n8n-nodes-base.jira",
  "typeVersion": 1,
  "position": [450, 300],
  "parameters": {
    "resource": "issue",
    "operation": "get",
    "issueKey": "={{ $json.ticket_id }}"
  },
  "credentials": {
    "jiraSoftwareCloudApi": {
      "id": "2",
      "name": "Jira Credentials"
    }
  }
}

Code Node

{
  "id": "code-1",
  "name": "Transform Data",
  "type": "n8n-nodes-base.code",
  "typeVersion": 2,
  "position": [650, 300],
  "parameters": {
    "jsCode": "const items = $input.all();\n\nreturn items.map(item => {\n  return {\n    json: {\n      processed: true,\n      data: item.json\n    }\n  };\n});"
  }
}

Respond to Webhook Node

{
  "id": "respond-1",
  "name": "Respond to Webhook",
  "type": "n8n-nodes-base.respondToWebhook",
  "typeVersion": 1.1,
  "position": [850, 300],
  "parameters": {
    "respondWith": "json",
    "responseBody": "={{ $json }}",
    "options": {
      "responseCode": 200,
      "responseHeaders": {
        "entries": [
          {
            "name": "Content-Type",
            "value": "application/json"
          }
        ]
      }
    }
  }
}

IF Node

{
  "id": "if-1",
  "name": "Check Condition",
  "type": "n8n-nodes-base.if",
  "typeVersion": 2,
  "position": [450, 300],
  "parameters": {
    "conditions": {
      "options": {
        "caseSensitive": true,
        "leftValue": "",
        "typeValidation": "strict"
      },
      "conditions": [
        {
          "id": "condition-1",
          "leftValue": "={{ $json.status }}",
          "rightValue": "active",
          "operator": {
            "type": "string",
            "operation": "equals"
          }
        }
      ],
      "combinator": "and"
    },
    "options": {}
  }
}

Connections Object

The connections object maps node outputs to inputs:

{
  "connections": {
    "Webhook": {
      "main": [
        [
          {
            "node": "Jira",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Jira": {
      "main": [
        [
          {
            "node": "Respond to Webhook",
            "type": "main",
            "index": 0
          }
        ]
      ]
    }
  }
}

Connection Structure

{
  "Source Node Name": {
    "main": [
      [
        {
          "node": "Target Node Name",
          "type": "main",
          "index": 0
        }
      ]
    ]
  }
}

• Key: Source node's display name
• main: Array of output ports (index 0 = first output)
• node: Target node's display name
• type: Connection type (usually "main")
• index: Target input port index

Multiple Outputs (IF Node)

{
  "Check Condition": {
    "main": [
      [
        {
          "node": "True Branch",
          "type": "main",
          "index": 0
        }
      ],
      [
        {
          "node": "False Branch",
          "type": "main",
          "index": 0
        }
      ]
    ]
  }
}

Multiple Connections from Same Output

{
  "Webhook": {
    "main": [
      [
        {
          "node": "Process A",
          "type": "main",
          "index": 0
        },
        {
          "node": "Process B",
          "type": "main",
          "index": 0
        }
      ]
    ]
  }
}

Settings Object

{
  "settings": {
    "executionOrder": "v1",
    "saveManualExecutions": true,
    "callerPolicy": "workflowsFromSameOwner",
    "errorWorkflow": "error-handler-workflow-id",
    "timezone": "America/New_York",
    "executionTimeout": 3600
  }
}

Settings Properties

Static Data

The staticData property stores persistent data across executions:

{
  "staticData": {
    "lastProcessedId": "12345",
    "counter": 42,
    "cache": {
      "key1": "value1"
    }
  }
}

Expressions

n8n uses expressions for dynamic values, wrapped in ={{ }}:

{
  "parameters": {
    "url": "https://api.example.com/users/{{ $json.userId }}",
    "body": "={{ JSON.stringify($json.data) }}",
    "headers": {
      "Authorization": "Bearer {{ $env.API_TOKEN }}"
    }
  }
}

Expression Variables

Complete Example

{
  "name": "Get Jira Ticket",
  "nodes": [
    {
      "id": "webhook-1",
      "name": "Webhook",
      "type": "n8n-nodes-base.webhook",
      "typeVersion": 2,
      "position": [250, 300],
      "webhookId": "get-jira-ticket",
      "parameters": {
        "path": "get-jira-ticket",
        "httpMethod": "POST",
        "responseMode": "responseNode",
        "options": {}
      }
    },
    {
      "id": "jira-1",
      "name": "Get Issue",
      "type": "n8n-nodes-base.jira",
      "typeVersion": 1,
      "position": [450, 300],
      "parameters": {
        "resource": "issue",
        "operation": "get",
        "issueKey": "={{ $json.body.ticket_id }}"
      },
      "credentials": {
        "jiraSoftwareCloudApi": {
          "id": "1",
          "name": "Jira API"
        }
      }
    },
    {
      "id": "respond-1",
      "name": "Respond",
      "type": "n8n-nodes-base.respondToWebhook",
      "typeVersion": 1.1,
      "position": [650, 300],
      "parameters": {
        "respondWith": "json",
        "responseBody": "={{ { success: true, data: $json } }}",
        "options": {}
      }
    }
  ],
  "connections": {
    "Webhook": {
      "main": [
        [
          {
            "node": "Get Issue",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Get Issue": {
      "main": [
        [
          {
            "node": "Respond",
            "type": "main",
            "index": 0
          }
        ]
      ]
    }
  },
  "settings": {
    "executionOrder": "v1"
  },
  "staticData": null
}

Credential References

Credentials are referenced by ID and name, but never include actual secrets:

{
  "credentials": {
    "jiraSoftwareCloudApi": {
      "id": "1",
      "name": "Jira Credentials"
    }
  }
}

After importing a workflow, credential references must be updated to match the target n8n instance's credential IDs.

Validation Rules

Node IDs

• Must be unique within the workflow
• Recommended format: UUID or descriptive slug

Node Names

• Must be unique within the workflow
• Used in connections object as keys
• Displayed on the canvas

Connections

• Source and target node names must exist in nodes array
• Output index must be valid for the source node type
• Input index must be valid for the target node type

TypeVersion

• Must match a valid version for the node type
• Newer typeVersions may have different parameter schemas

Python Workflow Builder

import json
import uuid
from dataclasses import dataclass, field
from typing import Dict, List, Any, Optional

@dataclass
class Node:
    """n8n workflow node."""
    name: str
    type: str
    type_version: float
    position: tuple[int, int]
    parameters: Dict[str, Any] = field(default_factory=dict)
    credentials: Optional[Dict[str, Any]] = None
    id: str = field(default_factory=lambda: str(uuid.uuid4()))

    def to_dict(self) -> dict:
        """Convert to n8n JSON format."""
        result = {
            "id": self.id,
            "name": self.name,
            "type": self.type,
            "typeVersion": self.type_version,
            "position": list(self.position),
            "parameters": self.parameters
        }
        if self.credentials:
            result["credentials"] = self.credentials
        return result

@dataclass
class Workflow:
    """n8n workflow builder."""
    name: str
    nodes: List[Node] = field(default_factory=list)
    connections: Dict[str, Any] = field(default_factory=dict)
    settings: Dict[str, Any] = field(
        default_factory=lambda: {"executionOrder": "v1"}
    )

    def add_node(self, node: Node) -> "Workflow":
        """Add a node to the workflow."""
        self.nodes.append(node)
        return self

    def connect(
        self,
        source: str,
        target: str,
        source_output: int = 0,
        target_input: int = 0
    ) -> "Workflow":
        """Connect two nodes."""
        if source not in self.connections:
            self.connections[source] = {"main": []}

        # Ensure enough output slots
        while len(self.connections[source]["main"]) <= source_output:
            self.connections[source]["main"].append([])

        self.connections[source]["main"][source_output].append({
            "node": target,
            "type": "main",
            "index": target_input
        })
        return self

    def to_dict(self) -> dict:
        """Convert to n8n JSON format."""
        return {
            "name": self.name,
            "nodes": [n.to_dict() for n in self.nodes],
            "connections": self.connections,
            "settings": self.settings,
            "staticData": None
        }

    def to_json(self, indent: int = 2) -> str:
        """Export as JSON string."""
        return json.dumps(self.to_dict(), indent=indent)


# Usage example
workflow = Workflow(name="Get Jira Ticket")

webhook = Node(
    name="Webhook",
    type="n8n-nodes-base.webhook",
    type_version=2,
    position=(250, 300),
    parameters={
        "path": "get-jira-ticket",
        "httpMethod": "POST",
        "responseMode": "responseNode"
    }
)

jira = Node(
    name="Get Issue",
    type="n8n-nodes-base.jira",
    type_version=1,
    position=(450, 300),
    parameters={
        "resource": "issue",
        "operation": "get",
        "issueKey": "={{ $json.body.ticket_id }}"
    },
    credentials={
        "jiraSoftwareCloudApi": {"id": "1", "name": "Jira API"}
    }
)

respond = Node(
    name="Respond",
    type="n8n-nodes-base.respondToWebhook",
    type_version=1.1,
    position=(650, 300),
    parameters={
        "respondWith": "json",
        "responseBody": "={{ { success: true, data: $json } }}"
    }
)

workflow.add_node(webhook).add_node(jira).add_node(respond)
workflow.connect("Webhook", "Get Issue")
workflow.connect("Get Issue", "Respond")

print(workflow.to_json())

Related Documentation

• n8n CLI Commands
• n8n REST API
• Common Patterns