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
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": """
Processed Data
Results
{{#each $json}}
{{name}} ({{age}}) - {{#if processed}}Processed{{else}}Pending{{/if}}
{{/each}}
""",
"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节点
Data Report
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; }
Data Processing Report
Generated on: {{$now}}
{{#each $json}}
{{name}}
ID: {{id}}
Age: {{age}}
Status: {{#if processed}}Processed{{else}}Pending{{/if}}
{{#if timestamp}}
Timestamp: {{timestamp}}
{{/if}}
{{/each}}
Summary
Total items: {{$json.length}}
Processed: {{#filter $json "processed" true}}{{/filter.length}}
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上创建问题