Claude Code + n8n — Integration Guide

User Request
    ↓
Claude Code (Brain)        ← Understands intent, generates workflow
    ↓
n8n Workflow (Engine)      ← Executes the workflow reliably
    ↓
External Services          ← APIs, databases, email, Slack, etc
    ↓
Results Back to User

Step 1: User sends request to Claude
"Create an email digest of Slack messages from #general over the past week"

Step 2: Claude generates n8n workflow JSON
{
  "nodes": [
    {
      "name": "Slack List Messages",
      "type": "n8n-nodes-base.slack",
      "parameters": {
        "channel": "general",
        "limit": 100
      }
    },
    {
      "name": "Filter Last Week",
      "type": "n8n-nodes-base.code",
      "parameters": {
        "jsCode": "return items.filter(msg => msg.ts > Math.floor(Date.now()/1000) - 604800)"
      }
    },
    {
      "name": "Email Digest",
      "type": "n8n-nodes-base.emailSend",
      "parameters": {
        "to": "[email protected]",
        "subject": "Weekly Digest",
        "body": "=new messages | map(...)"
      }
    }
  ]
}

Step 3: Claude uploads workflow to n8n via API
POST /api/v1/workflows
Authorization: Bearer [API_KEY]
Body: [workflow JSON]

Step 4: n8n stores workflow, ready to use
User can run it manually or schedule it

Step 5: n8n executes workflow, sends digest
Results saved to database, sent to user

# claude_workflow_generator.py
from anthropic import Anthropic

def generate_workflow(requirement: str, context: dict) -> dict:
    """Generate n8n workflow from natural language requirement"""

    client = Anthropic()

    # Provide Claude with n8n JSON schema
    system_prompt = """You are an n8n workflow generator.
    Given user requirements, generate valid n8n workflow JSON.

    Available nodes: Slack, Gmail, HTTP, Code, Spreadsheet, Database, etc.
    Format output as valid JSON that can be imported into n8n.
    Include proper error handling and data transformation.
    """

    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=4096,
        system=system_prompt,
        messages=[
            {
                "role": "user",
                "content": f"""Generate n8n workflow for:
                {requirement}

                Context:
                - n8n instance: {context.get('n8n_url')}
                - Available APIs: {', '.join(context.get('apis', []))}
                - Database: {context.get('database_type')}

                Output: Valid n8n workflow JSON only (no markdown, no explanation)"""
            }
        ]
    )

    # Parse JSON response
    import json
    workflow_json = json.loads(response.content[0].text)

    return workflow_json

# Usage
requirement = """
Monitor GitHub issues labeled 'bug'.
For each new issue, create a Slack message in #bugs.
Include issue title, description, and link.
"""

workflow = generate_workflow(requirement, {
    "n8n_url": "http://n8n.example.com",
    "apis": ["github", "slack"],
    "database_type": "postgresql"
})

# Upload to n8n
upload_workflow_to_n8n(workflow)

def review_and_optimize_workflow(workflow_json: dict) -> dict:
    """Have Claude review and optimize n8n workflow"""

    client = Anthropic()

    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=2048,
        messages=[
            {
                "role": "user",
                "content": f"""Review this n8n workflow and suggest optimizations:

{json.dumps(workflow_json, indent=2)}

Check for:
1. Unnecessary steps (can 2 nodes be combined?)
2. Missing error handling
3. Performance issues (batching, caching)
4. Security issues (hardcoded credentials, exposed data)
5. Cost optimization (API limits, expensive operations)

Output: Specific suggestions with code fixes"""
            }
        ]
    )

    suggestions = response.content[0].text
    print("Workflow Review:")
    print(suggestions)

    return workflow_json

Email arrives → n8n triggers Claude → Claude reads content → Claude generates response → n8n sends back

{
  "nodes": [
    {
      "name": "Email Trigger",
      "type": "n8n-nodes-base.webhookTrigger",
      "parameters": {
        "httpMethod": "POST",
        "path": "email-processor"
      }
    }
  ]
}

// In n8n Code node
const emailBody = $input.first().json.body;

// Call Claude API
const response = await fetch('https://api.anthropic.com/v1/messages', {
  method: 'POST',
  headers: {
    'x-api-key': process.env.ANTHROPIC_API_KEY,
    'content-type': 'application/json'
  },
  body: JSON.stringify({
    model: 'claude-3-5-sonnet-20241022',
    max_tokens: 1024,
    messages: [
      {
        role: 'user',
        content: `Email: ${emailBody}\n\nRespond with a helpful reply.`
      }
    ]
  })
});

const result = await response.json();
return { reply: result.content[0].text };

{
  "name": "Email Reply",
  "type": "n8n-nodes-base.emailSend",
  "parameters": {
    "to": "={{ $json.from }}",
    "subject": "Re: {{ $json.subject }}",
    "body": "={{ $json.reply }}"
  }
}

Raw content → Extract (Claude) → Transform (Code) → Publish (HTTP) → Store (DB)

# content-pipeline.json
nodes:
  - name: Get Content
    type: HTTP
    url: "https://api.example.com/articles"

  - name: Extract Key Points
    type: Code (Claude)
    prompt: "Summarize article in 3 bullet points"

  - name: Generate Social Post
    type: Code (Claude)
    prompt: "Create Twitter post for article"

  - name: Save to Database
    type: Database
    query: "INSERT INTO posts (content, summary, twitter_post)"

  - name: Post to Social
    type: Twitter
    action: "tweet"
    content: "={{ $json.twitter_post }}"

Code changes → Claude analyzes → Generate tests → n8n runs tests → Report results

def auto_test_workflow(code_diff: str) -> dict:
    """Generate and run tests for code changes"""

    client = Anthropic()

    # 1. Claude analyzes what changed
    analysis = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        messages=[
            {
                "role": "user",
                "content": f"What should be tested in this code change?\n\n{code_diff}"
            }
        ]
    )

    test_cases = analysis.content[0].text

    # 2. Claude generates test code
    generation = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=2048,
        messages=[
            {
                "role": "user",
                "content": f"Generate pytest tests for these cases:\n\n{test_cases}"
            }
        ]
    )

    test_code = generation.content[0].text

    # 3. n8n runs tests
    import subprocess
    result = subprocess.run(
        ["pytest", "-v"],
        input=test_code,
        capture_output=True,
        text=True
    )

    return {
        "test_code": test_code,
        "result": result.stdout,
        "passed": result.returncode == 0
    }

Support ticket → Claude categorizes → Route to specialist → Track resolution

{
  "name": "Ticket Received",
  "type": "Webhook"
}

{
  "name": "Classify Ticket",
  "type": "Code (Claude)",
  "prompt": "Classify: urgent/normal/low. Category: billing/technical/general"
}

{
  "name": "Route",
  "type": "Switch",
  "cases": [
    {"category": "billing", "assignee": "finance-team"},
    {"category": "technical", "assignee": "engineering"},
    {"category": "general", "assignee": "support"}
  ]
}

{
  "name": "Draft Response",
  "type": "Code (Claude)",
  "prompt": "Draft helpful response to customer"
}

{
  "name": "Send Reply",
  "type": "Email"
}

{
  "name": "Track SLA",
  "type": "Database",
  "query": "INSERT INTO tickets (id, category, status, response_sent_at)"
}

version: '3.8'

services:
  n8n:
    image: n8nio/n8n:latest
    container_name: n8n
    ports:
      - "5678:5678"
    environment:
      - N8N_BASIC_AUTH_ACTIVE=true
      - N8N_BASIC_AUTH_USER=admin
      - N8N_BASIC_AUTH_PASSWORD=${N8N_PASSWORD}
      - DB_TYPE=postgres
      - DB_POSTGRESDB_HOST=postgres
      - DB_POSTGRESDB_PORT=5432
      - DB_POSTGRESDB_DATABASE=n8n
      - DB_POSTGRESDB_USER=n8n
      - DB_POSTGRESDB_PASSWORD=${DB_PASSWORD}
    volumes:
      - n8n_data:/home/node/.n8n
    depends_on:
      - postgres

  postgres:
    image: postgres:15
    container_name: n8n-postgres
    environment:
      - POSTGRES_USER=n8n
      - POSTGRES_PASSWORD=${DB_PASSWORD}
      - POSTGRES_DB=n8n
    volumes:
      - postgres_data:/var/lib/postgresql/data

  claude-code-bridge:
    build: ./claude-code-bridge
    container_name: claude-bridge
    environment:
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
      - N8N_URL=http://n8n:5678
      - N8N_API_KEY=${N8N_API_KEY}
    ports:
      - "8000:8000"

volumes:
  n8n_data:
  postgres_data:

# claude-code-bridge/main.py
from fastapi import FastAPI, HTTPException
from anthropic import Anthropic
import httpx
import json

app = FastAPI()
client = Anthropic()

@app.post("/generate-workflow")
async def generate_workflow(request: dict):
    """Generate n8n workflow from Claude"""

    requirement = request.get("requirement")
    n8n_url = os.environ["N8N_URL"]
    api_key = os.environ["N8N_API_KEY"]

    # Claude generates workflow
    response = client.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=4096,
        messages=[
            {
                "role": "user",
                "content": f"Generate n8n workflow JSON for: {requirement}"
            }
        ]
    )

    workflow_json = json.loads(response.content[0].text)

    # Upload to n8n
    async with httpx.AsyncClient() as http_client:
        result = await http_client.post(
            f"{n8n_url}/api/v1/workflows",
            headers={"X-N8N-API-KEY": api_key},
            json=workflow_json
        )

    return {
        "workflow_id": result.json()["id"],
        "status": "created"
    }

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

# n8n credentials (never hardcode!)
{
  "credentials": {
    "slack_api_key": "{{ $env.SLACK_API_KEY }}",
    "github_token": "{{ $env.GITHUB_TOKEN }}",
    "database_url": "{{ $env.DATABASE_URL }}"
  }
}

# Prevent Claude from overwhelming APIs
class RateLimiter:
    def __init__(self, max_calls: int, time_window: int):
        self.max_calls = max_calls
        self.time_window = time_window
        self.calls = []

    def is_allowed(self) -> bool:
        now = time.time()
        self.calls = [c for c in self.calls if c > now - self.time_window]
        if len(self.calls) < self.max_calls:
            self.calls.append(now)
            return True
        return False

# Encrypt sensitive data in n8n
{
  "name": "Encrypt Data",
  "type": "Code",
  "code": """
  const crypto = require('crypto');
  const key = Buffer.from(process.env.ENCRYPTION_KEY);
  const iv = crypto.randomBytes(16);
  const cipher = crypto.createCipheriv('aes-256-cbc', key, iv);
  let encrypted = cipher.update(JSON.stringify($json), 'utf8', 'hex');
  encrypted += cipher.final('hex');
  return { encrypted, iv: iv.toString('hex') };
  """
}