DataOps-platform/docs/Task_Manager_MCP_说明.md

Task Manager MCP 工作流程说明

问题诊断

现象

Task Manager MCP 能够读取到任务，但是没有自动生成对应的代码。

原因分析

Task Manager MCP 的工作流程分为以下步骤：

1. 任务读取 ✅ - MCP 从数据库读取任务
2. 返回执行指令 ✅ - MCP 返回任务描述和执行指令
3. AI 生成代码 ❌ - 这一步需要 AI 主动执行
4. 更新任务状态 ❌ - AI 必须调用 MCP 工具更新状态

核心问题

MCP 只负责任务管理，不负责代码生成。代码生成需要 AI 来完成。

MCP 工作流程

1. 执行任务 (execute_task)

mcp_task-manager_execute_task({
  task_id: 7,
  auto_complete: true
})

返回内容:

• 任务信息（ID、名称、描述）
• 任务描述（包含 DDL、需求等）
• 执行指令（告诉 AI 如何生成代码）
• 更新状态的提醒（AI 必须调用 update_task_status）

注意: 这个函数不会自动生成代码，只是返回指令。

2. AI 生成代码

AI 需要：

1. 阅读任务描述
2. 理解需求
3. 生成 Python 代码
4. 保存到指定路径

3. 更新任务状态 (update_task_status)

mcp_task-manager_update_task_status({
  task_id: 7,
  status: "completed",
  code_name: "import_dept_data.py",
  code_path: "app/core/data_flow"
})

这一步是必须的，否则任务会一直停留在 "processing" 状态。

完整流程示例

任务 7: 导入数据资源的科室对照表

步骤 1: 读取任务

// 获取任务详情
mcp_task-manager_get_task_by_id({ task_id: 7 })

// 返回
{
  "task_id": 7,
  "task_name": "导入数据资源的科室对照表",
  "status": "pending",
  "task_description": "# Task: 导入数据资源的科室对照表\n..."
}

步骤 2: 执行任务

// 执行任务
mcp_task-manager_execute_task({ 
  task_id: 7, 
  auto_complete: true 
})

// 返回执行指令
{
  "message": "请生成代码并更新任务状态...",
  "task_description": "...",
  "steps": [
    "分析需求",
    "生成代码",
    "更新任务状态"
  ]
}

注意: 此时任务状态变为 "processing"。

步骤 3: AI 生成代码

AI 执行以下操作：

# 1. 创建代码文件
write("app/core/data_flow/import_dept_data.py", code_content)

# 2. 验证代码
python -m py_compile app/core/data_flow/import_dept_data.py

步骤 4: 更新任务状态

// AI 调用 MCP 工具更新状态
mcp_task-manager_update_task_status({
  task_id: 7,
  status: "completed",
  code_name: "import_dept_data.py",
  code_path: "app/core/data_flow"
})

// 返回
{
  "task_id": 7,
  "status": "completed",
  "update_time": "2025-11-28T10:33:37.833Z"
}

任务状态流转

pending (待处理)
    ↓
  execute_task 调用
    ↓
processing (处理中)
    ↓
  AI 生成代码
    ↓
  update_task_status 调用
    ↓
completed (已完成)

常见问题

Q1: 为什么任务一直停留在 "processing" 状态？

原因: AI 没有调用 update_task_status 更新状态。

解决方案: 在生成代码后，必须调用 update_task_status。

Q2: 任务状态是 "processing"，还能再次执行吗？

不能。只有 "pending" 状态的任务才能执行。

解决方案:

// 先更新为 pending
mcp_task-manager_update_task_status({
  task_id: 7,
  status: "pending"
})

// 再执行
mcp_task-manager_execute_task({ task_id: 7 })

Q3: MCP 会自动生成代码吗？

不会。MCP 只负责：

• 任务管理（读取、更新、删除）
• 返回执行指令

代码生成由 AI 负责。

Q4: 如何查看所有待处理的任务？

// 查看 pending 状态的任务
mcp_task-manager_get_pending_tasks()

// 查看所有任务
mcp_task-manager_get_all_tasks({ limit: 20 })

MCP 工具列表

工具	功能	参数
get_pending_tasks	获取待处理任务列表	无
get_task_by_id	根据ID获取任务详情	task_id
execute_task	执行任务（返回指令）	task_id, auto_complete
update_task_status	更新任务状态	task_id, status, code_name, code_path
process_all_tasks	批量处理所有待处理任务	auto_poll
create_task	创建新任务	task_name, task_description, create_by
get_all_tasks	获取所有任务（调试用）	limit

最佳实践

1. 自动化流程

// 1. 获取所有待处理任务
const pendingTasks = mcp_task-manager_get_pending_tasks()

// 2. 逐个执行
for (const task of pendingTasks) {
  // 执行任务（获取指令）
  const instructions = mcp_task-manager_execute_task({
    task_id: task.task_id,
    auto_complete: true
  })
  
  // AI 生成代码
  // ... (AI 操作)
  
  // 更新状态
  mcp_task-manager_update_task_status({
    task_id: task.task_id,
    status: "completed",
    code_name: "generated_file.py",
    code_path: "app/core/data_flow"
  })
}

2. 错误处理

try {
  // 执行任务
  const result = mcp_task-manager_execute_task({ task_id: 7 })
  
  // AI 生成代码
  // ...
  
  // 更新为完成
  mcp_task-manager_update_task_status({
    task_id: 7,
    status: "completed"
  })
} catch (error) {
  // 更新为失败
  mcp_task-manager_update_task_status({
    task_id: 7,
    status: "failed"
  })
}

3. 批量处理

// 启用自动轮询（每5分钟检查一次）
mcp_task-manager_process_all_tasks({
  auto_poll: true
})

任务描述格式

由 DataFlow 自动生成

当创建 DataFlow 时，系统会自动生成任务描述：

# Task: {任务名称}

## Data Source
- Type: postgresql
- Host: 10.52.31.104
- Port: 5432
- Database: hospital_db

## Source Tables (DDL)
```sql
CREATE TABLE source_table (...);

Target Tables (DDL)

CREATE TABLE target_table (
    ...,
    create_time TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

Update Mode

• Mode: Append (追加模式)

Request Content

{rule 字段的内容}

Implementation Steps

1. Extract data from source tables
2. Apply transformation logic
3. Write data to target table
4. Generate Python program

5. Generate n8n workflow

   
   ## 生成的代码结构
   
   ### 标准模板
   
   ```python
   """
   {任务名称}
   
   功能：{任务描述}
   模式：{update_mode}
   作者：cursor
   创建时间：{当前日期}
   """
   
   import logging
   from typing import Dict, List, Any
   from app.extensions import db
   from app.core.graph.neo4j_client import connect_graph
   
   logger = logging.getLogger(__name__)
   
   
   class DataProcessor:
   """数据处理器"""
       
   def __init__(self):
       self.processed_count = 0
       self.error_count = 0
       
   def get_data_source_info(self, bd_id: int):
       """获取数据源信息"""
       pass
       
   def extract_data(self):
       """提取数据"""
       pass
       
   def transform_data(self, data):
       """转换数据"""
       pass
       
   def load_data(self, data):
       """加载数据"""
       pass
       
   def run(self):
       """执行主流程"""
       pass
   
   
   def main_function():
   """主函数"""
   processor = DataProcessor()
   return processor.run()
   
   
   if __name__ == '__main__':
   result = main_function()
   print(f"处理结果: {result}")
   

与 DataFlow 集成

DataFlow 创建时自动生成任务

在 app/core/data_flow/dataflows.py 的 create_dataflow 函数中：

# 保存到 task_list 表
task_insert_sql = text("""
    INSERT INTO public.task_list 
    (task_name, task_description, status, code_name, code_path, create_by, create_time)
    VALUES 
    (:task_name, :task_description, :status, :code_name, :code_path, :create_by, :create_time)
""")

task_params = {
    'task_name': script_name,
    'task_description': task_description_md,  # Markdown 格式的任务描述
    'status': 'pending',
    'code_name': script_name,
    'code_path': 'app/core/data_flow',
    'create_by': 'cursor',
    'create_time': current_time
}

db.session.execute(task_insert_sql, task_params)

任务描述包含的信息

1. ✅ 数据源信息: 从 Neo4j 获取（如果有）
2. ✅ 源表 DDL: 根据 source_table IDs 生成
3. ✅ 目标表 DDL: 根据 target_table IDs 生成（含 create_time）
4. ✅ 更新模式: append 或 full
5. ✅ 规则说明: rule 字段内容
6. ✅ 实施步骤: 标准化步骤

总结

MCP 的职责

• ✅ 任务的 CRUD 操作
• ✅ 返回任务描述和执行指令
• ✅ 更新任务状态
• ❌ 不负责生成代码

AI 的职责

• ✅ 读取 MCP 返回的任务描述
• ✅ 理解需求并生成代码
• ✅ 保存代码到指定路径
• ✅ 必须调用 MCP 更新任务状态

关键要点

1. execute_task 不会自动生成代码，只返回指令
2. AI 必须主动生成代码
3. AI 必须调用 update_task_status 更新状态
4. 只有 "pending" 状态的任务才能执行
5. 任务描述由 DataFlow 自动生成，格式标准化

示例：任务 7 的完整流程

// 1. 查看任务
mcp_task-manager_get_task_by_id({ task_id: 7 })
// 状态: pending

// 2. 执行任务
mcp_task-manager_execute_task({ task_id: 7, auto_complete: true })
// 状态: processing
// 返回: 任务描述 + 执行指令

// 3. AI 生成代码
write("app/core/data_flow/import_dept_data.py", code)
// 文件已创建

// 4. 更新状态
mcp_task-manager_update_task_status({
  task_id: 7,
  status: "completed",
  code_name: "import_dept_data.py",
  code_path: "app/core/data_flow"
})
// 状态: completed ✅

相关文件

• app/core/data_flow/dataflows.py - DataFlow 创建时写入 task_list
• docs/DataFlow_task_list优化说明.md - 任务描述生成逻辑
• app/core/data_flow/import_dept_data.py - 任务 7 生成的代码

更新历史

• 2025-11-28: 诊断 Task Manager MCP 工作流程，明确 AI 和 MCP 的职责划分