# 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`)

```javascript
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`)

```javascript
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: 读取任务

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

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

#### 步骤 2: 执行任务

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

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

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

#### 步骤 3: AI 生成代码

AI 执行以下操作：

```python
# 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: 更新任务状态

```javascript
// 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" 状态的任务才能执行。

**解决方案**: 
```javascript
// 先更新为 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: 如何查看所有待处理的任务？

```javascript
// 查看 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. 自动化流程

```javascript
// 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. 错误处理

```javascript
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. 批量处理

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

## 任务描述格式

### 由 DataFlow 自动生成

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

```markdown
# 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)
```sql
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` 函数中：

```python
# 保存到 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 的完整流程

```javascript
// 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 的职责划分