n8n-flow-manager 🚀

n8n-flow-manager is a robust, production-ready Python SDK and CLI for the n8n automation platform. Unlike simple HTTP wrappers, this package is designed for DevOps workflows, providing type-safe models, async operations, workflow templating, and CI/CD integration capabilities.

✨ Features

Core Capabilities

⚡ Async-First Design: Built on httpx for high-performance async operations
🛡️ Type Safety: Complete Pydantic models for workflows, executions, and credentials
🔄 Smart Polling: Execute workflows and wait for completion with intelligent status checking
📝 Jinja2 Templating: Inject environment-specific variables into workflow definitions
🤖 Powerful CLI: Terminal commands for backup, deploy, sync, and manage workflows
🔐 Secure: API key authentication with proper error handling and retries
📦 Zero Config: Works with environment variables or explicit configuration

What Makes It Different?

Feature	n8n-flow-manager	Generic HTTP Clients
Type Validation	✅ Pydantic models	❌ Raw dicts
Async Support	✅ Native asyncio	⚠️ Sync only
Smart Execution	✅ run_and_wait()	❌ Manual polling
Templating	✅ Jinja2 built-in	❌ Not included
CLI Tools	✅ Full featured	❌ None
Error Handling	✅ Custom exceptions	⚠️ Generic errors

📂 Project Structure

n8n-flow-manager/
├── src/
│   └── n8n_manager/
│       ├── __init__.py           # Public API exports
│       ├── client.py             # Main async client
│       ├── exceptions.py         # Custom error types
│       ├── api/                  # API modules by resource
│       │   ├── workflows.py      # Workflow operations
│       │   ├── executions.py     # Execution management
│       │   └── credentials.py    # Credential handling
│       ├── models/               # Pydantic data models
│       │   ├── workflow.py       # Workflow structures
│       │   ├── execution.py      # Execution states
│       │   └── credential.py     # Credential types
│       ├── cli/                  # Command-line interface
│       │   └── main.py           # Typer CLI app
│       └── utils/                # Helper utilities
│           └── templating.py     # Jinja2 template engine
├── tests/                        # Pytest test suite
├── examples/                     # Usage examples
├── pyproject.toml                # Poetry configuration
└── README.md                     # This file

🚀 Installation

Requirements

Python 3.9 or higher
n8n instance (cloud or self-hosted)

Global CLI Installation (Recommended)

Install globally using pipx (the best practice for Python CLI tools):

# Install pipx if you don't have it
python3 -m pip install --user pipx
python3 -m pipx ensurepath

# Install n8n-flow-manager globally
pipx install n8n-flow-manager

# Verify installation
n8n-py --version
n8n-py --help

Note: If you don't have pipx, you can also use pip install --user n8n-flow-manager, but pipx is recommended as it isolates the package in its own environment.

Configuration

After installation, configure your n8n credentials using environment variables:

# Set your n8n credentials
export N8N_API_KEY="your_api_key_here"
export N8N_BASE_URL="https://your-instance.n8n.cloud"

# Test the connection
n8n-py health

Optional: To make credentials persistent, add them to your shell profile:

# For zsh (macOS default)
echo 'export N8N_API_KEY="your_api_key_here"' >> ~/.zshrc
echo 'export N8N_BASE_URL="https://your-instance.n8n.cloud"' >> ~/.zshrc
source ~/.zshrc

# For bash
echo 'export N8N_API_KEY="your_api_key_here"' >> ~/.bashrc
echo 'export N8N_BASE_URL="https://your-instance.n8n.cloud"' >> ~/.bashrc
source ~/.bashrc

Getting Your API Key

Open your n8n instance
Go to Settings → API
Click Create API Key
Copy the key and use it in the configuration above

That's it! You're ready to use n8n-flow-manager 🎉

For Developers

If you want to use n8n-flow-manager as a Python library in your projects:

# Install with pip
pip install n8n-flow-manager

# Use in your code
from n8n_manager import N8NClient

For contributing or development, see the Development section below.

🛠️ Usage Guide

1. Python SDK Usage

Basic Client Usage

import asyncio
from n8n_manager import N8NClient

async def main():
    # Initialize client (reads from environment)
    async with N8NClient() as client:

        # List all active workflows
        workflows = await client.workflows.list(active=True)
        for wf in workflows:
            print(f"Workflow: {wf.name} (ID: {wf.id})")

        # Get specific workflow
        workflow = await client.workflows.get("workflow_id")
        print(f"Nodes: {len(workflow.nodes)}")

        # Execute workflow and wait for result
        execution = await client.executions.run_and_wait(
            workflow_id="workflow_id",
            timeout=60
        )
        print(f"Status: {execution.status}")
        print(f"Success: {execution.is_successful}")

asyncio.run(main())

Creating Workflows Programmatically

from n8n_manager import N8NClient
from n8n_manager.models.workflow import Workflow, Node

async def create_simple_workflow():
    async with N8NClient() as client:
        workflow = Workflow(
            name="Python-Created Workflow",
            active=False,
            nodes=[
                Node(
                    name="Start",
                    type="n8n-nodes-base.start",
                    position=[250, 300],
                    parameters={}
                ),
                Node(
                    name="Set Data",
                    type="n8n-nodes-base.set",
                    position=[450, 300],
                    parameters={
                        "values": {
                            "string": [
                                {"name": "message", "value": "Hello from Python!"}
                            ]
                        }
                    }
                )
            ],
            connections={
                "Start": {
                    "main": [[{"node": "Set Data", "type": "main", "index": 0}]]
                }
            }
        )

        created = await client.workflows.create(workflow)
        print(f"Created workflow: {created.id}")

Using Templates

from n8n_manager.utils.templating import load_workflow_from_file

# Load workflow with template variables
workflow = load_workflow_from_file(
    "templates/data_sync.json",
    variables={
        "environment": "production",
        "api_endpoint": "https://api.example.com",
        "timeout": 30
    }
)

async with N8NClient() as client:
    deployed = await client.workflows.create(workflow)
    print(f"Deployed: {deployed.name}")

2. CLI Usage

The CLI provides powerful commands for workflow management.

Example Output:

$ n8n-py health
✓ Connection healthy!
  API URL: https://n8n.example.com/

$ n8n-py list-workflows
                              Workflows (33 found)
┏━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━┳━━━━━━━┓
┃ ID               ┃ Name                           ┃ Active ┃ Nodes ┃
┡━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━╇━━━━━━━┩
│ 1RDHBsmLkkTptybX │ Production Data Sync           │ ✓      │ 6     │
│ 2V3iCBkiOAPVzrUr │ Customer Onboarding            │ ✗      │ 7     │
│ 8qGqx5TW1QA7T8P9 │ Error Notifications            │ ✓      │ 2     │
└──────────────────┴────────────────────────────────┴────────┴───────┘

Commands

# List all workflows
n8n-py list-workflows

# List only active workflows
n8n-py list-workflows --active

Get Workflow Details

# Display workflow info
n8n-py get-workflow <workflow_id>

# Save workflow to file
n8n-py get-workflow <workflow_id> --output workflow.json

Deploy Workflows

# Deploy from JSON file
n8n-py deploy workflow.json

# Deploy with template variables
n8n-py deploy template.json --var environment=prod --var timeout=30

# Deploy and activate immediately
n8n-py deploy workflow.json --activate

Backup Workflows

# Backup all workflows
n8n-py backup --output ./backups

# Backup only active workflows
n8n-py backup --output ./backups --active-only

Execute Workflows

# Execute and wait for completion
n8n-py execute <workflow_id>

# Execute without waiting
n8n-py execute <workflow_id> --no-wait

# Execute with input data
n8n-py execute <workflow_id> --input data.json

Activate/Deactivate

# Activate workflow
n8n-py activate <workflow_id>

# Deactivate workflow
n8n-py deactivate <workflow_id>

Health Check

# Verify connection to n8n
n8n-py health

📚 Advanced Examples

CI/CD Integration

Use in GitHub Actions for automated deployments:

# .github/workflows/deploy-n8n.yml
name: Deploy n8n Workflows

on:
  push:
    branches: [main]

jobs:
  deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: '3.11'

      - name: Install n8n-flow-manager
        run: pip install -e .

      - name: Deploy workflows
        env:
          N8N_API_KEY: ${{ secrets.N8N_API_KEY }}
          N8N_BASE_URL: ${{ secrets.N8N_BASE_URL }}
        run: |
          n8n-py deploy workflows/production.json --activate

Environment-Specific Deployments

import asyncio
from n8n_manager import N8NClient
from n8n_manager.utils.templating import load_workflow_from_file

ENVIRONMENTS = {
    "dev": {
        "api_endpoint": "https://dev.api.example.com",
        "webhook_path": "webhook-dev",
        "timeout": 10
    },
    "prod": {
        "api_endpoint": "https://api.example.com",
        "webhook_path": "webhook",
        "timeout": 30
    }
}

async def deploy_to_environment(env: str):
    workflow = load_workflow_from_file(
        "templates/api_workflow.json",
        variables=ENVIRONMENTS[env]
    )

    async with N8NClient() as client:
        deployed = await client.workflows.create(workflow)
        await client.workflows.activate(deployed.id)
        print(f"Deployed to {env}: {deployed.id}")

# Deploy to production
asyncio.run(deploy_to_environment("prod"))

Monitoring and Logging

async def monitor_executions(workflow_id: str):
    async with N8NClient() as client:
        executions = await client.executions.list(
            workflow_id=workflow_id,
            limit=50
        )

        for execution in executions:
            if execution.is_failed:
                print(f"❌ Failed: {execution.id} at {execution.started_at}")
            elif execution.is_successful:
                print(f"✅ Success: {execution.id}")

🧪 Testing

Run the test suite with pytest:

# Install dev dependencies
poetry install --with dev

# Run all tests
poetry run pytest

# Run with coverage
poetry run pytest --cov=src/n8n_manager --cov-report=html

# Run specific test file
poetry run pytest tests/test_client.py

🔧 Development

Setting Up Development Environment

# Clone repository
git clone https://github.com/yourusername/n8n-flow-manager.git
cd n8n-flow-manager

# Install with dev dependencies
poetry install --with dev

# Install pre-commit hooks
poetry run pre-commit install

# Run linting
poetry run black src/ tests/
poetry run ruff src/ tests/

# Type checking
poetry run mypy src/

Project Roadmap

[x] Core client with async support
[x] Pydantic models for type safety
[x] Workflow, execution, and credential APIs
[x] CLI with Typer
[x] Jinja2 templating
[x] Smart execution polling
[ ] Webhook management API
[ ] Tag management
[ ] Bulk operations
[ ] Workflow validation before deploy
[ ] Integration tests with mock n8n server

📖 API Reference

N8NClient

Main client for interacting with n8n API.

Methods:

workflows - WorkflowAPI instance
executions - ExecutionAPI instance
credentials - CredentialAPI instance
health_check() - Verify API connection

WorkflowAPI

Methods:

list(active=None, tags=None) - List workflows
get(workflow_id) - Get workflow by ID
create(workflow) - Create new workflow
update(workflow_id, workflow) - Update workflow
delete(workflow_id) - Delete workflow
activate(workflow_id) - Activate workflow
deactivate(workflow_id) - Deactivate workflow

ExecutionAPI

Methods:

list(workflow_id=None, limit=100, status=None) - List executions
get(execution_id) - Get execution details
trigger_workflow(workflow_id, input_data=None) - Trigger execution
wait_for_execution(execution_id, timeout=300) - Wait for completion
run_and_wait(workflow_id, input_data=None, timeout=300) - Trigger and wait
retry(execution_id) - Retry failed execution
delete(execution_id) - Delete execution

CredentialAPI

Methods:

list(credential_type=None) - List credentials
get(credential_id) - Get credential by ID
create(credential) - Create credential
update(credential_id, credential) - Update credential
delete(credential_id) - Delete credential

🤝 Contributing

Contributions are welcome! Please follow these steps:

Fork the repository
Create a feature branch (git checkout -b feature/amazing-feature)
Commit your changes (git commit -m 'Add amazing feature')
Push to the branch (git push origin feature/amazing-feature)
Open a Pull Request

Contribution Guidelines

Write tests for new features
Follow existing code style (Black + Ruff)
Update documentation as needed
Add type hints to all functions
Keep commits atomic and well-described

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

🙏 Acknowledgments

n8n - The workflow automation platform
httpx - Async HTTP client
Pydantic - Data validation
Typer - CLI framework
Rich - Terminal formatting

📞 Support

Documentation: GitHub Wiki
Issues: GitHub Issues
Discussions: GitHub Discussions
n8n Community: n8n Community Forum

🎯 Use Cases

DevOps & CI/CD

Automate workflow deployments across environments
Version control your n8n workflows in Git
Integrate with GitLab/GitHub Actions pipelines

Disaster Recovery

Scheduled backups of all workflows
Quick restore capabilities
Environment replication

Multi-Tenant Management

Programmatically create workflows for new clients
Template-based workflow generation
Bulk operations across workflows

Monitoring & Observability

Track execution success rates
Monitor workflow health
Automated alerting on failures

Made with ❤️ for the n8n community

n8n-flow-manager

About n8n-flow-manager

Platforms

Languages

Links

README.md

n8n-flow-manager 🚀

✨ Features

Core Capabilities

What Makes It Different?

📂 Project Structure

🚀 Installation

Requirements

Global CLI Installation (Recommended)

Configuration

Getting Your API Key

For Developers

🛠️ Usage Guide

1. Python SDK Usage

Basic Client Usage

Creating Workflows Programmatically

Using Templates

2. CLI Usage

Commands

Get Workflow Details

Deploy Workflows

Backup Workflows

Execute Workflows

Activate/Deactivate

Health Check

📚 Advanced Examples

CI/CD Integration

Environment-Specific Deployments

Monitoring and Logging

🧪 Testing

🔧 Development

Setting Up Development Environment

Project Roadmap

📖 API Reference

N8NClient

WorkflowAPI

ExecutionAPI

CredentialAPI

🤝 Contributing

Contribution Guidelines

📄 License

🙏 Acknowledgments

📞 Support

🎯 Use Cases

DevOps & CI/CD

Disaster Recovery

Multi-Tenant Management

Monitoring & Observability