使用说明

N8N MCP服务器

一种模型上下文协议（MCP）服务器，提供与N8N工作流和执行交互的工具。此服务器允许您通过Cursor IDE以编程方式创建、管理和执行N8N工作流。

🚀 特性

🔗 连接管理：测试和验证N8N服务器连接
📋 工作流管理：创建、读取、更新、删除工作流
⚡ 工作流执行：使用自定义数据执行工作流
🎯 高级节点创建：使用所有可能的N8N节点创建工作流
💻 自定义代码支持：将自定义JavaScript、HTML和函数代码添加到任何节点
🔄 工作流生命周期：激活、停用和管理工作流状态
📊 执行跟踪：监控工作流执行和结果
🐳 容器化：已准备好使用Docker容器
🔧 交叉平台的：适用于Windows、macOS和Linux

🛠️ 可用工具

🔗 连接和基本操作

n8n_test_connection -测试与N8N服务器的连接
n8n_get_workflows -从N8N获取所有工作流
n8n_get_workflow -按ID获取特定工作流

📝 工作流创建

n8n_create_workflow -创建基本工作流
n8n_create_advanced_workflow -使用自定义节点和代码创建工作流
n8n_get_available_nodes -获取可用节点类型和模板的列表

🎯 节点管理

n8n_create_node_with_code -使用自定义代码创建节点

🔄 工作流生命周期

n8n_update_workflow -更新现有工作流
n8n_delete_workflow -删除工作流
n8n_activate_workflow -激活工作流
n8n_deactivate_workflow -停用工作流

⚡ 执行与监控

n8n_execute_workflow -使用自定义数据执行工作流
n8n_get_workflow_executions -获取工作流执行

🎯 支持的节点类型

MCP服务器支持创建具有以下节点类型的工作流：

🔧 核心节点

手动触发器 -手动工作流触发器
代码 -JavaScript代码执行
函数 -数据处理功能节点
httpRequest -HTTP API调用
网络钩子 Webhook 端点

📊 数据处理

集 -设置数据值
合并 -合并多个数据流
splitInBatches -将数据拆分为批次
如果 -条件逻辑
开关 -多路径路由

📤 输出节点

超文本标记语言 -生成HTML输出
标记语言 -生成Markdown输出
电子邮件 -发送电子邮件
文件操作 -文件操作

🔗 集成节点

googleSheets -Google表格集成
松弛 -Slack消息
不和谐 -不和谐消息
电报 -电报消息

⏰ 自动化节点

定时任务 -预定触发器

🚀 快速开始

先决条件

已安装Docker和Docker Compose
N8N实例正在运行且可访问
生成了N8N API密钥

步骤1：克隆和设置

# Clone the repository
git clone <repository-url>
cd n8n-mcp-server

# Run setup script
./setup.sh  # Linux/Mac
# OR
setup.bat   # Windows

步骤2：配置环境

编辑 .env 使用N8N配置的文件：

# N8N Configuration
N8N_BASE_URL=http://your-n8n-instance:5678
N8N_API_KEY=your-n8n-api-key-here
N8N_USERNAME=  # Optional
N8N_PASSWORD=  # Optional

步骤3：部署

# Start the container
docker-compose up -d

# Check status
docker-compose ps

# View logs
docker-compose logs n8n-mcp-server

步骤4：配置游标

添加到光标MCP配置(~/.cursor/mcp.json):

{
  "mcpServers": {
    "n8n": {
      "command": "docker",
      "args": ["run", "--rm", "-i", "n8n-mcp-server"],
      "env": {
        "N8N_BASE_URL": "http://your-n8n-instance:5678",
        "N8N_API_KEY": "your-n8n-api-key-here"
      }
    }
  }
}

📖 使用示例

🔗 测试连接

# Test if N8N server is accessible
result = await n8n_test_connection()
print(result)

📋 获取可用节点

# Get all available node types
nodes = await n8n_get_available_nodes()
print(nodes)

🎯 使用自定义代码创建节点

# Create a Code node with custom JavaScript
node_data = await n8n_create_node_with_code(
    node_type="code",
    node_name="Data Processor",
    custom_code="""
    return items.map(item => ({
        json: {
            ...item.json,
            processed: true,
            timestamp: new Date().toISOString()
        }
    }));
    """
)

📝 创建高级工作流

# Create a complex workflow with multiple nodes
workflow_config = {
    "name": "Data Processing Pipeline",
    "nodes_config": [
        {
            "type": "manualTrigger",
            "name": "Start",
            "position": [240, 300]
        },
        {
            "type": "code",
            "name": "Data Generator",
            "code": """
            return [
                { json: { id: 1, name: "Alice", age: 30 } },
                { json: { id: 2, name: "Bob", age: 25 } }
            ];
            """,
            "position": [460, 300]
        },
        {
            "type": "function",
            "name": "Process Data",
            "code": """
            for (const item of items) {
                item.json.processed = true;
                item.json.timestamp = new Date().toISOString();
            }
            return items;
            """,
            "position": [680, 300]
        },
        {
            "type": "html",
            "name": "HTML Output",
            "code": """
            <html>
            <head><title>Processed Data</title></head>
            <body>
                <h1>Results</h1>
                <ul>
                {{#each $json}}
                    <li>{{name}} ({{age}}) - {{#if processed}}Processed{{else}}Pending{{/if}}</li>
                {{/each}}
                </ul>
            </body>
            </html>
            """,
            "position": [900, 300]
        }
    ]
}

result = await n8n_create_advanced_workflow(
    name=workflow_config["name"],
    nodes_config=json.dumps(workflow_config["nodes_config"])
)

⚡ 执行工作流

# Execute a workflow with custom data
execution_data = {
    "input": "test data",
    "timestamp": "2024-01-01T00:00:00Z"
}

result = await n8n_execute_workflow(
    workflow_id="your-workflow-id",
    data=json.dumps(execution_data)
)

🔄 管理工作流生命周期

# Activate a workflow
await n8n_activate_workflow(workflow_id="your-workflow-id")

# Execute the workflow
await n8n_execute_workflow(workflow_id="your-workflow-id")

# Get execution history
executions = await n8n_get_workflow_executions(
    workflow_id="your-workflow-id",
    limit=10
)

# Deactivate the workflow
await n8n_deactivate_workflow(workflow_id="your-workflow-id")

🎨 节点代码示例

JavaScript代码节点

// Generate data
return [
    { json: { id: 1, name: "Alice", age: 30 } },
    { json: { id: 2, name: "Bob", age: 25 } },
    { json: { id: 3, name: "Charlie", age: 35 } }
];

// Process incoming data
for (const item of items) {
    item.json.processed = true;
    item.json.timestamp = new Date().toISOString();
    item.json.fullName = `${item.json.firstName} ${item.json.lastName}`;
}
return items;

// API response processing
const results = [];
for (const item of items) {
    if (item.json.status === 'success') {
        results.push({
            json: {
                id: item.json.id,
                message: 'Processed successfully',
                data: item.json.data
            }
        });
    }
}
return results;

功能节点

// Data transformation
for (const item of items) {
    // Add computed fields
    item.json.fullName = `${item.json.firstName} ${item.json.lastName}`;
    item.json.age = new Date().getFullYear() - item.json.birthYear;
    
    // Add processing timestamp
    item.json.processedAt = new Date().toISOString();
    
    // Add status based on conditions
    if (item.json.age >= 18) {
        item.json.status = 'adult';
    } else {
        item.json.status = 'minor';
    }
}
return items;

// Data filtering
const filteredItems = items.filter(item => 
    item.json.status === 'active' && 
    item.json.score > 80
);

return filteredItems.map(item => ({
    json: {
        id: item.json.id,
        name: item.json.name,
        score: item.json.score,
        category: item.json.score > 90 ? 'excellent' : 'good'
    }
}));

HTML节点

<html>
<head>
    <title>Data Report</title>
    <style>
        body { font-family: Arial, sans-serif; margin: 20px; }
        .header { background: #f0f0f0; padding: 10px; border-radius: 5px; }
        .item { margin: 10px 0; padding: 10px; border: 1px solid #ddd; }
        .success { background: #d4edda; }
        .error { background: #f8d7da; }
    </style>
</head>
<body>
    <div class="header">
        <h1>Data Processing Report</h1>
        <p>Generated on: {{$now}}</p>
    </div>
    
    {{#each $json}}
    <div class="item {{#if processed}}success{{else}}error{{/if}}">
        <h3>{{name}}</h3>
        <p><strong>ID:</strong> {{id}}</p>
        <p><strong>Age:</strong> {{age}}</p>
        <p><strong>Status:</strong> {{#if processed}}Processed{{else}}Pending{{/if}}</p>
        {{#if timestamp}}
        <p><strong>Timestamp:</strong> {{timestamp}}</p>
        {{/if}}
    </div>
    {{/each}}
    
    <div class="header">
        <h2>Summary</h2>
        <p>Total items: {{$json.length}}</p>
        <p>Processed: {{#filter $json "processed" true}}{{/filter.length}}</p>
    </div>
</body>
</html>

HTTP请求节点

// URL: https://api.example.com/users
// Method: GET
// Headers: { "Authorization": "Bearer {{$json.token}}" }

// For POST requests with data
// URL: https://api.example.com/users
// Method: POST
// Body: { "name": "{{$json.name}}", "email": "{{$json.email}}" }

🔧 发展

地方发展

# Install dependencies
pip install -r requirements.txt

# Set environment variables
export N8N_BASE_URL=http://localhost:5678
export N8N_API_KEY=your-api-key

# Run the server
python n8n_mcp_server_simple.py

测试

# Run the test suite
python test_container.py

# Test specific functionality
python -c "
import asyncio
from n8n_mcp_server_simple import n8n_client
result = asyncio.run(n8n_client.test_connection())
print(result)
"

🔍 故障排除

常见问题

连接失败
- 验证N8N是否在正确的端口上运行
- 检查API密钥是否有效
- 确保网络连接
工作流创建失败
- 验证是否支持节点类型
- 检查自定义代码中的JSON语法
- 确保提供所需参数
执行错误
- 检查工作流是否已激活
- 验证输入数据格式
- 查看执行日志

调试模式

通过设置环境变量启用调试日志记录：

export DEBUG=1
export PYTHONUNBUFFERED=1

日志

# View container logs
docker-compose logs n8n-mcp-server

# Follow logs in real-time
docker-compose logs -f n8n-mcp-server

# View specific log levels
docker-compose logs n8n-mcp-server | grep ERROR

🔒 安全

最佳实践

环境变量：永远不要硬编码机密
API密钥：定期旋转按键
网络安全：使用适当的网络隔离
集装箱安全：以最低权限运行

生产部署

对于生产环境，使用机密管理：

# Use Docker secrets
docker run -d --name n8n-mcp-server \
  --secret n8n_api_key \
  -e N8N_API_KEY_FILE=/run/secrets/n8n_api_key \
  n8n-mcp-server

📊 监控

健康检查

容器包括健康检查端点：

# Check container health
docker inspect n8n-mcp-server | grep Health -A 10

指标

监控工作流执行和性能：

# Get execution metrics
executions = await n8n_get_workflow_executions(
    workflow_id="your-workflow-id",
    limit=100
)

# Analyze execution data
for execution in executions['data']:
    print(f"Execution {execution['id']}: {execution['status']}")

🤝 贡献

分叉存储库
创建要素分支
进行更改
添加测试
提交拉取请求

📄 许可证

此项目根据MIT许可证获得许可-有关详细信息，请参阅许可证文件。

🆘 支持

对于问题和疑问：

检查故障排除部分
查看N8N文档了解API详细信息
在GitHub上创建问题