n8n+MCP强强联合：让AI助手真正掌控你的自动化工作流

# 创建n8n数据存储目录
mkdir -p ~/.n8n

# 启动n8n容器
docker run -d \
  --name n8n \
  -p 5678:5678 \
  -v ~/.n8n:/home/node/.n8n \
  n8nio/n8n

# 通过npm全局安装
npm install n8n -g

# 启动n8n
n8n start

# 克隆仓库
git clone https://github.com/czlonkowski/n8n-mcp.git

# 进入项目目录
cd n8n-mcp

# 安装依赖
npm install

# 构建项目
npm run build

# 在你的项目中安装
npm install n8n-mcp

# n8n服务器地址
N8N_BASE_URL=http://localhost:5678

# n8n API访问令牌
N8N_API_KEY=your_n8n_api_key_here

# MCP服务器端口
MCP_SERVER_PORT=3000

# 日志级别（可选）
LOG_LEVEL=info

# 开发模式（带热重载）
npm run dev

# 生产模式
npm start

[INFO] MCP Server running on port 3000
[INFO] Connected to n8n at http://localhost:5678
[INFO] Available workflows: 15

// 导入n8n-mcp客户端
const { N8nMcpClient } = require('n8n-mcp');

async function testConnection() {
  // 初始化客户端
  const client = new N8nMcpClient({
    baseUrl: process.env.N8N_BASE_URL,
    apiKey: process.env.N8N_API_KEY
  });

  try {
    // 测试连接
    const status = await client.healthCheck();
    console.log('连接状态:', status);

    // 获取工作流列表
    const workflows = await client.listWorkflows();
    console.log('可用工作流数量:', workflows.length);

    // 列出工作流名称
    workflows.forEach(wf => {
      console.log(`- ${wf.name} (ID: ${wf.id})`);
    });

  } catch (error) {
    console.error('连接失败:', error.message);
  }
}

testConnection();

node test.js

{
  "name": "trigger_workflow",
  "description": "触发指定n8n工作流的执行",
  "parameters": {
    "type": "object",
    "properties": {
      "workflow_id": {
        "type": "string",
        "description": "工作流的唯一标识符"
      },
      "input_data": {
        "type": "object",
        "description": "传递给工作流的输入数据"
      }
    },
    "required": ["workflow_id"]
  }
}

// 获取所有工作流的详细信息
const workflows = await client.listWorkflows({
  includeInactive: false  // 是否包含已禁用的工作流
});

// 工作流对象包含以下字段：
// - id: 唯一标识符
// - name: 工作流名称
// - active: 是否处于激活状态
// - createdAt: 创建时间
// - updatedAt: 最后更新时间

// 根据ID获取工作流详细信息
const workflow = await client.getWorkflow('workflow_id_here');

// 返回的工作流对象包含完整的节点配置
console.log('工作流节点数:', workflow.nodes.length);
console.log('工作流连接:', workflow.connections);

// 通过代码创建新工作流
const newWorkflow = await client.createWorkflow({
  name: '我的自动化工作流',
  nodes: [
    {
      name: 'Webhook',
      type: 'n8n-nodes-base.webhook',
      parameters: {},
      position: [250, 300]
    },
    {
      name: 'HTTP Request',
      type: 'n8n-nodes-base.httpRequest',
      parameters: {
        method: 'GET',
        url: 'https://api.example.com/data'
      },
      position: [450, 300]
    }
  ],
  connections: {
    'Webhook': {
      'main': [[{
        'node': 'HTTP Request',
        'type': 'main',
        'index': 0
      }]]
    }
  }
});

// 触发工作流执行
const execution = await client.executeWorkflow('workflow_id_here', {
  // 输入数据，会作为webhook或初始节点的输入
  userId: '12345',
  action: 'process_order',
  orderData: {
    items: [
      { name: '产品A', quantity: 2 },
      { name: '产品B', quantity: 1 }
    ],
    totalAmount: 299.99
  }
});

console.log('执行ID:', execution.id);
console.log('执行状态:', execution.status);

// 查询执行结果
const result = await client.getExecutionResult('execution_id_here');

console.log('执行状态:', result.status);
console.log('输出数据:', result.data);
console.log('执行耗时:', result.executionTime, 'ms');

// result.data包含工作流最后节点的输出
// 可以遍历查看每个节点的执行结果
if (result.data.resultData) {
  Object.keys(result.data.resultData).forEach(nodeName => {
    console.log(`节点 ${nodeName} 输出:`, result.data.resultData[nodeName]);
  });
}

// 同步执行，等待结果返回
const result = await client.executeWorkflowAndWait('workflow_id_here', {
  data: 'your input data'
}, {
  timeout: 60000,  // 超时时间（毫秒）
  pollInterval: 1000  // 轮询间隔（毫秒）
});

console.log('最终结果:', result);

// 订阅工作流执行事件
client.subscribe('workflow.execution.started', (event) => {
  console.log('工作流开始执行:', event.workflowId);
});

client.subscribe('workflow.execution.completed', (event) => {
  console.log('工作流执行完成:', {
    workflowId: event.workflowId,
    executionId: event.executionId,
    status: event.status,
    duration: event.duration
  });
});

client.subscribe('workflow.execution.failed', (event) => {
  console.error('工作流执行失败:', {
    workflowId: event.workflowId,
    error: event.error
  });
});

// 获取工作流的执行历史
const history = await client.getExecutionHistory('workflow_id_here', {
  limit: 20,  // 返回数量限制
  includeData: true  // 是否包含执行数据
});

history.forEach(exec => {
  const status = exec.status === 'success' ? '✓' : '✗';
  console.log(`${status} ${exec.id} - ${exec.startedAt} (${exec.executionTime}ms)`);
});

// 更新工作流的某个节点
await client.updateWorkflowNode('workflow_id_here', 'node_name', {
  parameters: {
    url: 'https://new-api-endpoint.com/data',
    method: 'POST'
  }
});

// 添加新节点到现有工作流
await client.addWorkflowNode('workflow_id_here', {
  name: 'New Node',
  type: 'n8n-nodes-base.httpRequest',
  parameters: {
    method: 'GET',
    url: 'https://example.com/api'
  },
  position: [650, 300]
});

// 配置webhook节点
{
  "path": "send-custom-email",
  "responseMode": "responseNode",
  "options": {}
}

// 设置邮件内容
{
  "mode": "manual",
  "assignments": {
    "assignments": [
      {
        "name": "to",
        "value": "={{ $json.body.to }}"
      },
      {
        "name": "subject",
        "value": "={{ $json.body.subject }}"
      },
      {
        "name": "content",
        "value": "={{ $json.body.content }}"
      }
    ]
  },
  "options": {}
}

// 配置邮件发送
{
  "fromEmail": "noreply@yourcompany.com",
  "to": "={{ $json.to }}",
  "subject": "={{ $json.subject }}",
  "text": "={{ $json.content }}",
  "emailFormat": "html"
}

/**
 * AI邮件自动化助手
 * 
 * 这个模块展示了如何将n8n-mcp与AI助手结合，
 * 实现智能化的邮件发送功能
 */

const { N8nMcpClient } = require('n8n-mcp');

class EmailAutomationAssistant {
  constructor(config) {
    this.client = new N8nMcpClient(config);
    this.webhookBaseUrl = config.webhookBaseUrl;
  }

  /**
   * 初始化连接
   */
  async initialize() {
    console.log('正在连接到n8n...');
    const status = await this.client.healthCheck();
    console.log('连接成功! n8n版本:', status.version);

    // 获取邮件工作流
    const workflows = await this.client.listWorkflows();
    this.emailWorkflow = workflows.find(w => 
      w.name.includes('email') || w.name.includes('邮件')
    );

    if (!this.emailWorkflow) {
      throw new Error('未找到邮件工作流，请先在n8n中创建');
    }

    console.log('邮件工作流已就绪:', this.emailWorkflow.name);
  }

  /**
   * 发送简单邮件
   * 
   * @param {string} to - 收件人邮箱
   * @param {string} subject - 邮件主题
   * @param {string} content - 邮件内容
   */
  async sendSimpleEmail(to, subject, content) {
    try {
      const result = await this.client.executeWorkflow(
        this.emailWorkflow.id,
        {
          to,
          subject,
          content,
          timestamp: new Date().toISOString()
        }
      );

      console.log('邮件发送任务已提交');
      console.log('执行ID:', result.id);

      return {
        success: true,
        executionId: result.id
      };
    } catch (error) {
      console.error('发送失败:', error.message);
      return {
        success: false,
        error: error.message
      };
    }
  }

  /**
   * 发送批量邮件
   * 
   * @param {Array} recipients - 收件人列表
   * @param {string} subject - 邮件主题
   * @param {Function} contentGenerator - 内容生成函数
   */
  async sendBatchEmails(recipients, subject, contentGenerator) {
    const results = [];

    for (const recipient of recipients) {
      const content = await contentGenerator(recipient);

      const result = await this.sendSimpleEmail(
        recipient.email,
        subject,
        content
      );

      results.push({
        recipient: recipient.email,
        ...result
      });

      // 避免发送过快
      await this.delay(1000);
    }

    return {
      total: recipients.length,
      success: results.filter(r => r.success).length,
      failed: results.filter(r => !r.success).length,
      details: results
    };
  }

  /**
   * AI驱动的智能邮件发送
   * 根据用户指令自动处理邮件发送
   * 
   * @param {string} instruction - AI指令
   */
  async handleAIInstruction(instruction) {
    // 模拟AI解析指令
    const parsed = this.parseInstruction(instruction);

    if (parsed.action === 'send_email') {
      return await this.sendSimpleEmail(
        parsed.to,
        parsed.subject,
        parsed.content
      );
    }

    if (parsed.action === 'send_batch') {
      return await this.sendBatchEmails(
        parsed.recipients,
        parsed.subject,
        parsed.contentGenerator
      );
    }

    return { error: '无法识别的指令' };
  }

  /**
   * 解析AI指令（简化版本）
   * 实际应用中，这里会接入真正的AI模型
   */
  parseInstruction(instruction) {
    // 简化实现，实际应使用AI模型解析
    const lower = instruction.toLowerCase();

    if (lower.includes('发送') && lower.includes('邮件')) {
      // 提取收件人、主题、内容（实际应用中由AI提取）
      return {
        action: 'send_email',
        to: 'recipient@example.com',
        subject: '自动发送的邮件',
        content: '这是由AI助手自动生成的邮件内容'
      };
    }

    throw new Error('无法解析指令');
  }

  // 工具方法：延时
  delay(ms) {
    return new Promise(resolve => setTimeout(resolve, ms));
  }
}

// 主程序
async function main() {
  const assistant = new EmailAutomationAssistant({
    baseUrl: process.env.N8N_BASE_URL,
    apiKey: process.env.N8N_API_KEY,
    webhookBaseUrl: process.env.WEBHOOK_BASE_URL
  });

  try {
    // 初始化
    await assistant.initialize();

    // 示例1：发送简单邮件
    console.log('\n--- 示例1：发送简单邮件 ---');
    const simpleResult = await assistant.sendSimpleEmail(
      'user@example.com',
      '测试邮件',
      '<h1>这是一封测试邮件</h1><p>来自n8n-mcp自动化系统</p>'
    );
    console.log('结果:', simpleResult);

    // 示例2：批量发送
    console.log('\n--- 示例2：批量发送邮件 ---');
    const batchRecipients = [
      { email: 'user1@example.com', name: '用户1' },
      { email: 'user2@example.com', name: '用户2' },
      { email: 'user3@example.com', name: '用户3' }
    ];

    const batchResult = await assistant.sendBatchEmails(
      batchRecipients,
      '批量测试邮件',
      (recipient) => `亲爱的 ${recipient.name}，这是一封个性化邮件！`
    );
    console.log('批量发送结果:', batchResult);

  } catch (error) {
    console.error('程序执行错误:', error);
  }
}

// 运行主程序
main();

node email-automation.js

正在连接到n8n...
连接成功! n8n版本: 1.0.0
邮件工作流已就绪: Email Automation Workflow

--- 示例1：发送简单邮件 ---
邮件发送任务已提交
执行ID: abc123-def456
结果: { success: true, executionId: 'abc123-def456' }

--- 示例2：批量发送邮件 ---
批量发送结果: { total: 3, success: 3, failed: 0, details: [...] }

{
  "path": "data-query",
  "httpMethod": "POST"
}

// 这个节点模拟数据查询，实际可能连接数据库或API
function processQuery($input, $vars) {
  const query = $input.body.query;
  const params = $input.body.params || {};

  // 模拟数据分析逻辑
  const results = {
    query: query,
    results: generateSampleData(query, params),
    timestamp: new Date().toISOString(),
    recordCount: 10
  };

  return results;
}

// 生成示例数据（实际应用中替换为真实数据查询）
function generateSampleData(query, params) {
  const data = [];
  const count = params.limit || 10;

  for (let i = 0; i < count; i++) {
    data.push({
      id: i + 1,
      value: Math.random() * 1000,
      category: ['A', 'B', 'C'][i % 3],
      date: new Date(Date.now() - i * 86400000).toISOString()
    });
  }

  return data;
}

return processQuery($input, $vars);

{
  "model": "gpt-4",
  "prompt": "分析以下数据，用简洁的语言总结关键洞察：\n{{ $json.results }}",
  "temperature": 0.3
}

/**
 * AI数据分析助手
 * 
 * 允许用户使用自然语言查询数据，
 * 系统自动调用后端工作流进行分析
 */

const { N8nMcpClient } = require('n8n-mcp');

class DataAnalysisAssistant {
  constructor(config) {
    this.client = new N8nMcpClient(config);
  }

  async initialize() {
    await this.client.healthCheck();
    console.log('数据助手初始化完成');

    // 查找数据查询工作流
    const workflows = await this.client.listWorkflows();
    this.queryWorkflow = workflows.find(w => 
      w.name.toLowerCase().includes('data') ||
      w.name.toLowerCase().includes('query') ||
      w.name.toLowerCase().includes('分析')
    );

    if (!this.queryWorkflow) {
      throw new Error('未找到数据查询工作流');
    }
  }

  /**
   * 执行数据查询
   * 
   * @param {string} naturalQuery - 自然语言查询
   * @param {Object} options - 查询选项
   */
  async query(naturalQuery, options = {}) {
    console.log('处理查询:', naturalQuery);

    // 解析自然语言为结构化查询
    const structuredQuery = this.nlpToQuery(naturalQuery);

    // 执行工作流
    const execution = await this.client.executeWorkflow(
      this.queryWorkflow.id,
      {
        query: structuredQuery.query,
        params: {
          limit: options.limit || 10,
          startDate: structuredQuery.dateRange?.start,
          endDate: structuredQuery.dateRange?.end,
          filters: structuredQuery.filters
        },
        originalQuery: naturalQuery
      }
    );

    // 等待结果
    const result = await this.client.waitForExecution(
      execution.id,
      { timeout: 30000 }
    );

    return this.formatResults(result, naturalQuery);
  }

  /**
   * 简单的NLP解析（简化版本）
   * 实际应用中应接入专门的NLP服务
   */
  nlpToQuery(naturalQuery) {
    const query = {};
    const lower = naturalQuery.toLowerCase();

    // 提取日期范围
    if (lower.includes('昨天')) {
      const yesterday = new Date();
      yesterday.setDate(yesterday.getDate() - 1);
      query.dateRange = {
        start: this.startOfDay(yesterday),
        end: this.endOfDay(yesterday)
      };
    }

    if (lower.includes('本周')) {
      const now = new Date();
      query.dateRange = {
        start: this.startOfWeek(now),
        end: this.endOfWeek(now)
      };
    }

    // 提取数量限制
    const limitMatch = naturalQuery.match(/前(\d+)条|显示(\d+)条|取(\d+)条/);
    if (limitMatch) {
      query.limit = parseInt(limitMatch[1] || limitMatch[2] || limitMatch[3]);
    }

    // 提取筛选条件
    query.filters = {};
    if (lower.includes('类别A')) {
      query.filters.category = 'A';
    }
    if (lower.includes('类别B')) {
      query.filters.category = 'B';
    }

    // 生成最终查询字符串
    query.query = this.buildQueryString(query);

    return query;
  }

  /**
   * 构建查询字符串
   */
  buildQueryString(parsedQuery) {
    let query = 'SELECT * FROM data';
    const conditions = [];

    if (parsedQuery.dateRange) {
      conditions.push(
        `date >= '${parsedQuery.dateRange.start}' AND date <= '${parsedQuery.dateRange.end}'`
      );
    }

    if (parsedQuery.filters) {
      Object.entries(parsedQuery.filters).forEach(([key, value]) => {
        conditions.push(`${key} = '${value}'`);
      });
    }

    if (conditions.length > 0) {
      query += ' WHERE ' + conditions.join(' AND ');
    }

    if (parsedQuery.limit) {
      query += ` LIMIT ${parsedQuery.limit}`;
    }

    return query;
  }

  /**
   * 格式化结果
   */
  formatResults(rawResults, originalQuery) {
    return {
      originalQuery,
      query: rawResults.query,
      recordCount: rawResults.recordCount,
      timestamp: rawResults.timestamp,
      data: rawResults.results,
      summary: this.generateSummary(rawResults)
    };
  }

  /**
   * 生成数据摘要
   */
  generateSummary(data) {
    if (!data.results || data.results.length === 0) {
      return '没有找到匹配的数据';
    }

    const count = data.results.length;
    const avg = data.results.reduce((sum, r) => sum + r.value, 0) / count;
    const max = Math.max(...data.results.map(r => r.value));
    const min = Math.min(...data.results.map(r => r.value));

    return `共 ${count} 条记录，数值范围 ${min.toFixed(2)} - ${max.toFixed(2)}，平均值 ${avg.toFixed(2)}`;
  }

  // 日期工具方法
  startOfDay(date) {
    return new Date(date.setHours(0, 0, 0, 0)).toISOString();
  }

  endOfDay(date) {
    return new Date(date.setHours(23, 59, 59, 999)).toISOString();
  }

  startOfWeek(date) {
    const d = new Date(date);
    const day = d.getDay();
    const diff = d.getDate() - day + (day === 0 ? -6 : 1);
    return this.startOfDay(new Date(d.setDate(diff)));
  }

  endOfWeek(date) {
    const start = new Date(this.startOfWeek(date));
    start.setDate(start.getDate() + 6);
    return this.endOfDay(start);
  }
}

// 演示
async function demo() {
  const assistant = new DataAnalysisAssistant({
    baseUrl: process.env.N8N_BASE_URL,
    apiKey: process.env.N8N_API_KEY
  });

  await assistant.initialize();

  // 示例查询
  const queries = [
    '查询昨天前10条数据',
    '本周类别A的数据',
    '显示最新的5条记录'
  ];

  for (const q of queries) {
    console.log(`\n处理查询: "${q}"`);
    const result = await assistant.query(q);
    console.log('结果:', JSON.stringify(result, null, 2));
  }
}

demo();

┌─────────────────────────────────────────────────────────┐
│                    AI 调度器                             │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────────┐  │
│  │ 任务解析器   │→ │ 优先级评估   │→ │ 调度决策引擎    │  │
│  └─────────────┘  └─────────────┘  └─────────────────┘  │
└──────────────────────────┬──────────────────────────────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────────┐
│                   n8n-mcp 层                             │
│  ┌─────────────────────────────────────────────────────┐│
│  │  任务队列管理器  │  工作流注册表  │  执行状态监控   ││
│  └─────────────────────────────────────────────────────┘│
└──────────────────────────┬──────────────────────────────┘
                           │
                           ▼
┌─────────────────────────────────────────────────────────┐
│                      n8n 引擎                            │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐ │
│  │ 工作流1  │  │ 工作流2  │  │ 工作流3  │  │ 工作流N  │ │
│  └──────────┘  └──────────┘  └──────────┘  └──────────┘ │
└─────────────────────────────────────────────────────────┘

/**
 * AI智能任务调度系统
 * 
 * 这个系统可以根据任务描述自动：
 * 1. 解析任务类型和参数
 * 2. 评估任务优先级
 * 3. 选择合适的工作流执行
 * 4. 监控执行状态并反馈结果
 */

const { N8nMcpClient } = require('n8n-mcp');

/**
 * 任务优先级枚举
 */
const Priority = {
  CRITICAL: 1,  // 紧急重要
  HIGH: 2,      // 重要
  NORMAL: 3,    // 普通
  LOW: 4        // 低
};

/**
 * 任务状态枚举
 */
const TaskStatus = {
  PENDING: 'pending',
  RUNNING: 'running',
  COMPLETED: 'completed',
  FAILED: 'failed',
  CANCELLED: 'cancelled'
};

/**
 * 任务定义类
 */
class Task {
  constructor(id, type, description, params) {
    this.id = id;
    this.type = type;
    this.description = description;
    this.params = params;
    this.priority = Priority.NORMAL;
    this.status = TaskStatus.PENDING;
    this.createdAt = new Date();
    this.startedAt = null;
    this.completedAt = null;
    this.result = null;
    this.error = null;
  }
}

/**
 * AI任务调度器主类
 */
class AITaskScheduler {
  constructor(config) {
    this.client = new N8nMcpClient(config);
    this.tasks = new Map();
    this.workflowRegistry = new Map();
    this.eventHandlers = new Map();
  }

  /**
   * 初始化调度器
   */
  async initialize() {
    console.log('正在初始化AI任务调度器...');

    // 检查n8n连接
    const status = await this.client.healthCheck();
    console.log('n8n连接状态:', status.status);

    // 加载工作流注册表
    await this.loadWorkflowRegistry();

    // 启动任务处理循环
    this.startProcessingLoop();

    console.log('调度器初始化完成');
    console.log('已注册工作流:', this.workflowRegistry.size);
  }

  /**
   * 加载工作流注册表
   * 将n8n中的工作流映射为可调度的任务类型
   */
  async loadWorkflowRegistry() {
    const workflows = await this.client.listWorkflows({ includeInactive: false });

    workflows.forEach(wf => {
      // 根据工作流名称自动分类
      const category = this.categorizeWorkflow(wf.name);
      this.workflowRegistry.set(category, {
        id: wf.id,
        name: wf.name,
        category: category,
        active: wf.active,
        avgExecutionTime: 0  // 可后续添加统计
      });
    });

    // 注册默认的工作流映射
    this.registerDefaultMappings();
  }

  /**
   * 注册默认的工作流映射
   */
  registerDefaultMappings() {
    // 如果没有找到对应的工作流，可以使用这些默认处理
    const defaults = {
      'email': { handler: this.handleEmailTask.bind(this) },
      'data': { handler: this.handleDataTask.bind(this) },
      'report': { handler: this.handleReportTask.bind(this) },
      'notification': { handler: this.handleNotificationTask.bind(this) },
      'sync': { handler: this.handleSyncTask.bind(this) }
    };

    Object.entries(defaults).forEach(([key, value]) => {
      if (!this.workflowRegistry.has(key)) {
        this.workflowRegistry.set(key, {
          name: `Default ${key} handler`,
          category: key,
          handler: value.handler
        });
      }
    });
  }

  /**
   * 根据名称自动分类工作流
   */
  categorizeWorkflow(name) {
    const lower = name.toLowerCase();

    if (lower.includes('email') || lower.includes('邮件')) return 'email';
    if (lower.includes('data') || lower.includes('数据')) return 'data';
    if (lower.includes('report') || lower.includes('报告')) return 'report';
    if (lower.includes('notify') || lower.includes('通知')) return 'notification';
    if (lower.includes('sync') || lower.includes('同步')) return 'sync';

    return 'general';
  }

  /**
   * 接收AI指令并创建任务
   * 
   * @param {string} instruction - AI给出的指令
   * @returns {Task} 创建的任务对象
   */
  async createTaskFromInstruction(instruction) {
    // 解析指令
    const parsed = this.parseInstruction(instruction);

    // 创建任务
    const task = new Task(
      this.generateTaskId(),
      parsed.type,
      instruction,
      parsed.params
    );

    // 评估优先级
    task.priority = this.evaluatePriority(parsed);

    // 存储任务
    this.tasks.set(task.id, task);

    // 触发事件
    this.emit('task.created', task);

    console.log(`任务已创建: [${task.id}] ${task.type} (优先级: ${task.priority})`);

    return task;
  }

  /**
   * 解析AI指令
   * 实际应用中应接入AI模型
   */
  parseInstruction(instruction) {
    const result = {
      type: 'general',
      params: {}
    };

    const lower = instruction.toLowerCase();

    // 识别任务类型
    if (lower.includes('发送邮件') || lower.includes('发邮件')) {
      result.type = 'email';
      // 提取邮件相关参数
      const toMatch = instruction.match(/给(.+?)发|至(.+?)[,，]/);
      if (toMatch) result.params.to = toMatch[1] || toMatch[2];
    }

    if (lower.includes('查询') || lower.includes('获取数据')) {
      result.type = 'data';
    }

    if (lower.includes('报告') || lower.includes('汇总')) {
      result.type = 'report';
    }

    if (lower.includes('通知') || lower.includes('提醒')) {
      result.type = 'notification';
    }

    // 识别急切程度
    if (lower.includes('紧急') || lower.includes('立即') || lower.includes('马上')) {
      result.urgency = 'high';
    }

    return result;
  }

  /**
   * 评估任务优先级
   */
  evaluatePriority(parsed) {
    // 紧急程度高
    if (parsed.urgency === 'high') {
      return Priority.CRITICAL;
    }

    // 根据任务类型调整
    switch (parsed.type) {
      case 'notification':
        return Priority.HIGH;
      case 'email':
        return Priority.NORMAL;
      case 'report':
        return Priority.LOW;
      default:
        return Priority.NORMAL;
    }
  }

  /**
   * 执行任务
   */
  async executeTask(taskId) {
    const task = this.tasks.get(taskId);
    if (!task) {
      throw new Error(`任务不存在: ${taskId}`);
    }

    console.log(`开始执行任务: [${task.id}]`);

    task.status = TaskStatus.RUNNING;
    task.startedAt = new Date();
    this.emit('task.started', task);

    try {
      // 获取对应的工作流
      const workflow = this.workflowRegistry.get(task.type);

      if (!workflow) {
        throw new Error(`未找到处理类型 '${task.type}' 的工作流`);
      }

      // 如果有处理器函数，直接调用
      if (workflow.handler) {
        task.result = await workflow.handler(task);
      } else {
        // 否则通过n8n-mcp执行
        const execution = await this.client.executeWorkflow(
          workflow.id,
          task.params
        );
        task.result = await this.client.waitForExecution(execution.id);
      }

      task.status = TaskStatus.COMPLETED;
      task.completedAt = new Date();

      this.emit('task.completed', task);
      console.log(`任务完成: [${task.id}]`);

    } catch (error) {
      task.status = TaskStatus.FAILED;
      task.error = error.message;
      task.completedAt = new Date();

      this.emit('task.failed', { task, error });
      console.error(`任务失败: [${task.id}] - ${error.message}`);
    }

    return task;
  }

  /**
   * 任务处理循环
   */
  startProcessingLoop() {
    this.processingInterval = setInterval(async () => {
      await this.processPendingTasks();
    }, 1000);  // 每秒检查一次
  }

  /**
   * 处理待执行的任务
   */
  async processPendingTasks() {
    // 获取所有待执行的任务
    const pendingTasks = Array.from(this.tasks.values())
      .filter(t => t.status === TaskStatus.PENDING)
      .sort((a, b) => a.priority - b.priority);  // 按优先级排序

    // 只处理最高优先级的任务
    if (pendingTasks.length > 0) {
      await this.executeTask(pendingTasks[0].id);
    }
  }

  /**
   * 获取任务状态
   */
  getTaskStatus(taskId) {
    const task = this.tasks.get(taskId);
    if (!task) return null;

    return {
      id: task.id,
      type: task.type,
      status: task.status,
      priority: task.priority,
      createdAt: task.createdAt,
      duration: task.startedAt 
        ? (task.completedAt - task.startedAt) / 1000 + 's' 
        : null,
      result: task.result,
      error: task.error
    };
  }

  /**
   * 获取调度器统计信息
   */
  getStats() {
    const tasks = Array.from(this.tasks.values());

    return {
      total: tasks.length,
      pending: tasks.filter(t => t.status === TaskStatus.PENDING).length,
      running: tasks.filter(t => t.status === TaskStatus.RUNNING).length,
      completed: tasks.filter(t => t.status === TaskStatus.COMPLETED).length,
      failed: tasks.filter(t => t.status === TaskStatus.FAILED).length,
      registeredWorkflows: this.workflowRegistry.size
    };
  }

  // 事件系统
  on(event, handler) {
    if (!this.eventHandlers.has(event)) {
      this.eventHandlers.set(event, []);
    }
    this.eventHandlers.get(event).push(handler);
  }

  emit(event, data) {
    const handlers = this.eventHandlers.get(event) || [];
    handlers.forEach(handler => handler(data));
  }

  // 工具方法
  generateTaskId() {
    return `task_${Date.now()}_${Math.random().toString(36).substr(2, 9)}`;
  }

  /**
   * 关闭调度器
   */
  shutdown() {
    if (this.processingInterval) {
      clearInterval(this.processingInterval);
    }
    console.log('调度器已关闭');
  }

  // 任务类型处理器
  async handleEmailTask(task) {
    console.log('处理邮件任务...');
    // 邮件处理逻辑
    return { sent: true };
  }

  async handleDataTask(task) {
    console.log('处理数据任务...');
    // 数据处理逻辑
    return { queried: true };
  }

  async handleReportTask(task) {
    console.log('处理报告任务...');
    // 报告生成逻辑
    return { generated: true };
  }

  async handleNotificationTask(task) {
    console.log('处理通知任务...');
    // 通知发送逻辑
    return { notified: true };
  }

  async handleSyncTask(task) {
    console.log('处理同步任务...');
    // 数据同步逻辑
    return { synced: true };
  }
}

// 演示程序
async function demo() {
  const scheduler = new AITaskScheduler({
    baseUrl: process.env.N8N_BASE_URL,
    apiKey: process.env.N8N_API_KEY
  });

  // 注册事件监听
  scheduler.on('task.created', task => {
    console.log(`[事件] 任务创建: ${task.id}`);
  });

  scheduler.on('task.started', task => {
    console.log(`[事件] 任务开始: ${task.id}`);
  });

  scheduler.on('task.completed', task => {
    console.log(`[事件] 任务完成: ${task.id}`);
  });

  scheduler.on('task.failed', ({ task, error }) => {
    console.log(`[事件] 任务失败: ${task.id} - ${error}`);
  });

  try {
    // 初始化
    await scheduler.initialize();

    // 模拟AI指令
    const instructions = [
      '立即给张三发送一封邮件，内容是明天的会议提醒',
      '查询本月的销售数据并生成汇总报告',
      '同步所有客户信息到CRM系统'
    ];

    // 创建任务
    console.log('\n--- 创建任务 ---');
    const tasks = [];
    for (const instruction of instructions) {
      const task = await scheduler.createTaskFromInstruction(instruction);
      tasks.push(task);
    }

    // 等待任务执行
    console.log('\n--- 等待任务执行 ---');
    await new Promise(resolve => setTimeout(resolve, 3000));

    // 查看结果
    console.log('\n--- 任务结果 ---');
    tasks.forEach(task => {
      const status = scheduler.getTaskStatus(task.id);
      console.log(`${task.id}: ${status.status}`);
    });

    // 查看统计
    console.log('\n--- 调度器统计 ---');
    console.log(scheduler.getStats());

  } finally {
    scheduler.shutdown();
  }
}

// 运行演示
demo();

// 客服工作流配置
const customerServiceWorkflow = {
  name: '智能客服系统',
  nodes: [
    {
      name: 'Webhook入口',
      type: 'n8n-nodes-base.webhook',
      parameters: {
        path: 'customer-service',
        responseMode: 'lastNode'
      }
    },
    {
      name: 'AI意图识别',
      type: '@n8n/n8n-nodes-langchain.agent',
      parameters: {
        model: 'gpt-4',
        systemPrompt: `
          你是一个客服助手，需要判断客户的问题类型：
          1. 如果是简单咨询，直接回答
          2. 如果需要操作数据，使用tools
          3. 如果无法回答，转人工
        `,
        tools: ['query_order', 'query_product', 'create_ticket']
      }
    },
    {
      name: '工单创建',
      type: 'n8n-nodes-base.httpRequest',
      parameters: {
        method: 'POST',
        url: 'https://helpdesk-api.example.com/tickets',
        body: '={{ {customer: $json.customer, issue: $json.issue} }}'
      }
    }
  ]
};

/**
 * 数据同步任务配置
 */

class DataSyncScheduler {
  constructor(mcpClient) {
    this.client = mcpClient;
    this.syncTasks = new Map();
  }

  /**
   * 配置同步任务
   */
  async setupSyncTask(taskName, config) {
    const workflow = await this.client.createWorkflow({
      name: `数据同步 - ${taskName}`,
      nodes: [
        // 源数据节点
        {
          name: '数据源',
          type: 'n8n-nodes-base.httpRequest',
          parameters: {
            method: 'GET',
            url: config.source.url,
            authentication: 'genericCredentialType',
            genericAuthType: config.source.authType
          }
        },
        // 数据转换节点
        {
          name: '数据清洗',
          type: 'n8n-nodes-base.code',
          parameters: {
            jsCode: `
              const data = $input.all();
              // 数据清洗逻辑
              const cleaned = data.map(item => ({
                id: item.json.id,
                value: item.json.value,
                timestamp: new Date().toISOString(),
                synced: true
              }));
              return cleaned.map(d => ({ json: d }));
            `
          }
        },
        // 目标存储节点
        {
          name: '数据仓库',
          type: 'n8n-nodes-base.postgres',
          parameters: {
            operation: 'insert',
            table: config.target.table,
            columns: 'id, value, timestamp, synced'
          }
        }
      ]
    });

    this.syncTasks.set(taskName, {
      workflowId: workflow.id,
      schedule: config.schedule,
      status: 'active'
    });

    return workflow;
  }

  /**
   * 执行同步任务
   */
  async executeSync(taskName, options = {}) {
    const task = this.syncTasks.get(taskName);
    if (!task) {
      throw new Error(`同步任务不存在: ${taskName}`);
    }

    console.log(`开始同步: ${taskName}`);

    const execution = await this.client.executeWorkflow(
      task.workflowId,
      {
        syncMode: options.incremental ? 'incremental' : 'full',
        lastSyncTime: options.lastSyncTime
      }
    );

    return {
      taskName,
      executionId: execution.id,
      startedAt: new Date()
    };
  }
}

/**
 * 报表自动化系统
 */

class ReportAutomation {
  constructor(mcpClient) {
    this.client = mcpClient;
  }

  /**
   * 创建报表生成工作流
   */
  async createReportWorkflow(reportConfig) {
    const workflow = await this.client.createWorkflow({
      name: `报表 - ${reportConfig.name}`,
      nodes: [
        {
          name: '定时触发',
          type: 'n8n-nodes-base.schedule',
          parameters: {
            rule: {
              interval: [reportConfig.schedule]
            }
          }
        },
        {
          name: '数据查询',
          type: 'n8n-nodes-base.postgres',
          parameters: {
            operation: 'executeQuery',
            query: reportConfig.query
          }
        },
        {
          name: '报表生成',
          type: 'n8n-nodes-base.code',
          parameters: {
            jsCode: `
              const data = $input.all();
              const yesterday = new Date();
              yesterday.setDate(yesterday.getDate() - 1);

              return [{
                json: {
                  reportDate: yesterday.toISOString().split('T')[0],
                  totalRecords: data.length,
                  summary: calculateSummary(data),
                  details: data
                }
              }];

              function calculateSummary(data) {
                return {
                  count: data.length,
                  sum: data.reduce((s, d) => s + d.json.value, 0),
                  avg: data.reduce((s, d) => s + d.json.value, 0) / data.length
                };
              }
            `
          }
        },
        {
          name: '邮件发送',
          type: 'n8n-nodes-base.email',
          parameters: {
            to: reportConfig.recipients.join(','),
            subject: `={{ $json.reportDate }} ${reportConfig.name}`,
            body: `={{ $json.summary }}`
          }
        }
      ]
    });

    // 激活工作流
    await this.client.activateWorkflow(workflow.id);

    return workflow;
  }

  /**
   * 手动触发报表生成
   */
  async generateReport(workflowId, dateRange) {
    const execution = await this.client.executeWorkflow(workflowId, {
      dateRange: dateRange,
      manualTrigger: true
    });

    // 等待执行完成
    const result = await this.client.waitForExecution(execution.id, {
      timeout: 60000
    });

    return result;
  }
}