python-n8n

A lightweight, fully-typed Python SDK for the n8n REST API (v1.1.1).

• Three runtime dependencies — httpx + pydantic + pyyaml, nothing else.
• Async-first, sync-friendly — AsyncN8nClient for async code, N8nClient for sync.
• Complete coverage — all 59 endpoints across 10 resource groups.
• Type-safe — every parameter, request body, and response is a Pydantic v2 model.
• Pagination built-in — cursor-based pagination handled automatically.
• Schema drift detection — verify your SDK version matches your n8n instance.

Requires Python 3.10+.

Installation

From source:

git clone https://github.com/sirghiny/n8n-python-client.git
cd n8n-client
pip install -e ".[dev]"

From a .whl file:

pip install python_n8n-<version>-py3-none-any.whl

Quick Start

from n8n import N8nClient, Tag

client = N8nClient(
    base_url="https://my-n8n.example.com",
    api_key="your-api-key",
)

# Create a tag
tag = client.tags.create(Tag(name="production"))
print(tag.id)

# List all workflows
for wf in client.workflows.list_all(active=True):
    print(wf.id, wf.name)

Async

from n8n import AsyncN8nClient

async with AsyncN8nClient(base_url="https://my-n8n.example.com", api_key="key") as client:
    workflows = await client.workflows.list_all(active=True).collect()
    print(f"Found {len(workflows)} active workflows")

Environment Variables

Instead of passing base_url and api_key directly, you can set:

Variable	Maps to
N8N_BASE_URL	base_url
N8N_API_KEY	api_key

# These are read automatically if not provided explicitly
client = N8nClient()

Resources

All endpoints are accessed through resource namespaces on the client:

Resource	Methods
client.audit	generate
client.credentials	list, create, update, delete, get_schema, transfer
client.data_tables	list, create, get, update, delete, list_rows, insert_rows, update_rows, upsert_row, delete_rows
client.executions	list, get, delete, retry, stop, stop_many, get_tags, update_tags
client.projects	list, create, update, delete, list_users, add_users, remove_user, change_user_role
client.source_control	pull
client.tags	list, create, get, update, delete
client.users	list, create, get, delete, change_role
client.variables	list, create, update, delete
client.workflows	list, create, get, update, delete, activate, deactivate, get_version, transfer, get_tags, update_tags

Every resource with a list method also has a list_all method that auto-paginates.

Pagination

# Single page
page = client.workflows.list(limit=10)
print(page.data)         # list[Workflow]
print(page.next_cursor)  # str | None

# Auto-paginate — iterates across all pages
for wf in client.workflows.list_all(active=True):
    print(wf.name)

# Collect everything into a list
all_wfs = await async_client.workflows.list_all().collect()

Error Handling

The SDK maps HTTP status codes to typed exceptions:

from n8n.exceptions import NotFoundError, AuthenticationError, N8nError

try:
    wf = client.workflows.get("nonexistent")
except NotFoundError:
    print("Workflow not found")
except AuthenticationError:
    print("Invalid API key")
except N8nError as e:
    print(f"API error {e.status_code}: {e.message}")
    print(e.response)  # raw httpx.Response for debugging

Exception	Status Code
BadRequestError	400
AuthenticationError	401
ForbiddenError	403
NotFoundError	404
ConflictError	409
ServerError	500+
ConnectionError	Network failures

WorkflowBuilder

Building n8n's nested connections dict by hand is tedious. WorkflowBuilder provides a fluent API:

from n8n import N8nClient
from n8n.builder import WorkflowBuilder

wf = (
    WorkflowBuilder("Webhook to Slack")
    .add_node(
        name="Webhook",
        type="n8n-nodes-base.webhook",
        type_version=2,
        position=(0, 0),
        parameters={"path": "incoming", "httpMethod": "POST"},
    )
    .add_node(
        name="Slack",
        type="n8n-nodes-base.slack",
        type_version=2,
        position=(200, 0),
        parameters={"channel": "#alerts", "text": "New event received"},
        credentials={"slackOAuth2Api": {"id": "12", "name": "Slack Prod"}},
    )
    .connect("Webhook", "Slack")
    .settings(timezone="UTC")
    .build()  # returns a Workflow model — no API call
)

# Submit it
client = N8nClient(base_url="...", api_key="...")
created = client.workflows.create(wf)
client.workflows.activate(created.id)

Multi-output nodes

For nodes like If or Switch with multiple outputs, specify source_output:

builder = (
    WorkflowBuilder("Branching")
    .add_node(name="If", type="n8n-nodes-base.if", type_version=2, position=(0, 0), parameters={})
    .add_node(name="True", type="n8n-nodes-base.set", type_version=3, position=(200, -80), parameters={})
    .add_node(name="False", type="n8n-nodes-base.set", type_version=3, position=(200, 80), parameters={})
    .connect("If", "True", source_output=0)
    .connect("If", "False", source_output=1)
)

Validation

build() validates the graph and raises WorkflowBuildError if:

• There are no nodes
• A connection references a node that doesn't exist
• Duplicate node names are used

Orphan nodes (no connections) produce a warning.

Schema Drift Detection

Check if your SDK version matches a running n8n instance:

# Against a local spec file
python -m n8n --file ./latest-openapi.yml

# Against a remote URL
python -m n8n --url https://docs.n8n.io/api/v1/openapi.yml

Programmatic usage:

from n8n import check_compatibility

report = check_compatibility(remote_spec_path="./latest-openapi.yml")
print(report.is_compatible)       # bool
print(report.added_endpoints)     # endpoints in remote but not in SDK
print(report.removed_endpoints)   # endpoints in SDK but not in remote
print(report.summary())           # human-readable diff

Models

All models use snake_case in Python and serialize to camelCase for the API. Both forms are accepted when constructing models:

from n8n.models import Workflow, WorkflowSettings

# camelCase from API responses works
wf = Workflow.model_validate({"createdAt": "2024-01-01T00:00:00Z", "name": "Test"})

# snake_case in Python code works too
wf = Workflow(name="Test", settings=WorkflowSettings(timezone="UTC"))

# Serializes to camelCase for the API
wf.model_dump(by_alias=True, exclude_unset=True)
# {"name": "Test", "settings": {"timezone": "UTC"}}

All models use extra="allow", so new fields added by future n8n versions won't break your code — they'll be accessible via model.model_extra.

Date-time fields are kept as str (ISO 8601). Use datetime.fromisoformat() if you need datetime objects.

Contributing

See CONTRIBUTING.md for development setup and contribution guidelines.

Development

Setup

# Using pyenv (recommended)
make pyenv

# Or manually
pip install -e ".[dev]"

Commands

make test          # unit tests (no Docker needed)
make e2e           # e2e tests (requires Docker n8n)
make lint          # ruff check
make format        # ruff format
make typecheck     # mypy --strict
make check         # lint + typecheck + all tests

E2E Tests

E2E tests run against a real n8n instance:

# Start n8n
docker compose up -d

# Set credentials in .env
# N8N_BASE_URL=http://localhost:5678
# N8N_API_KEY=your-api-key

# Run e2e tests
make e2e

License

MIT