MCPlanManager

Instructions

Implementation Reference

• src/mcplanmanager/app.py:148-164 (handler)

  MCP tool handler for 'editDependencies', decorated with @mcp.tool() for automatic registration, processes Pydantic inputs and delegates to PlanManager method.

  def editDependencies(edits: List[DependencyEdit]) -> ToolResponse[dict]:
      """
      以批量、事务性的方式编辑一个或多个任务的依赖关系。
  
      此工具允许 'set' 或 'update' 操作，所有编辑将在应用前进行全面验证。
      如果任何指令失败，整个操作将回滚。
  
      Args:
          edits (List[DependencyEdit]): 一个包含编辑指令对象的列表，每个对象的结构如下:
            - task_id (int): 要修改的任务ID。
            - action (Literal["set", "update"]): 要执行的操作。
            - dependencies (Optional[List[int]]): 当 action 为 'set' 时，提供新的完整依赖ID列表。
            - add (Optional[List[int]]): 当 action 为 'update' 时，提供要添加的依赖ID列表。
            - remove (Optional[List[int]]): 当 action 为 'update' 时，提供要移除的依赖ID列表。
      """
      edit_dicts = [edit.model_dump(exclude_none=True) for edit in edits]
      return plan_manager.edit_dependencies_in_batch(edit_dicts)

• src/mcplanmanager/models.py:23-31 (schema)

  Pydantic BaseModel defining the input schema for each edit operation in the editDependencies tool.

  class DependencyEdit(BaseModel):
      """
      用于editDependencies工具，定义单个依赖编辑操作的模型。
      """
      task_id: int = Field(..., description="要修改的任务ID。")
      action: Literal["set", "update"] = Field(..., description="要执行的操作：'set' 或 'update'。")
      dependencies: Optional[List[int]] = Field(default=None, description="当action为'set'时，提供新的完整依赖ID列表。")
      add: Optional[List[int]] = Field(default=None, description="当action为'update'时，提供要添加的依赖ID列表。")
      remove: Optional[List[int]] = Field(default=None, description="当action为'update'时，提供要移除的依赖ID列表。")

• src/mcplanmanager/plan_manager.py:445-509 (helper)

  Core batch dependency editing logic in PlanManager, including validation of tasks/dependencies, update operations (set/update), circular dependency detection, and transactional application.

  def edit_dependencies_in_batch(self, edits: List[Dict]) -> Dict:
      """
      以批量方式编辑多个任务的依赖关系
      该操作是事务性的：所有编辑指令在应用前都会被验证。
      """
      try:
          # --- 验证阶段 ---
          original_tasks_copy = deepcopy(self.plan_data["tasks"])
          temp_tasks_map = {task['id']: task for task in original_tasks_copy}
          
          for edit in edits:
              task_id = edit.get("task_id")
              action = edit.get("action")
  
              if task_id is None or action is None:
                  raise ValueError("Each edit must contain 'task_id' and 'action'")
  
              task_to_edit = temp_tasks_map.get(task_id)
              if not task_to_edit:
                  raise ValueError(f"Task {task_id} not found in plan")
  
              if action == "set":
                  new_deps = edit.get("dependencies", [])
                  for dep_id in new_deps:
                      if dep_id not in temp_tasks_map:
                          raise ValueError(f"Dependency task {dep_id} not found")
                  task_to_edit["dependencies"] = new_deps
  
              elif action == "update":
                  add_deps = edit.get("add", [])
                  remove_deps = edit.get("remove", [])
                  
                  current_deps_set = set(task_to_edit["dependencies"])
                  
                  for dep_id in add_deps:
                      if dep_id not in temp_tasks_map:
                          raise ValueError(f"Dependency task to add ({dep_id}) not found")
                      current_deps_set.add(dep_id)
                  
                  current_deps_set.difference_update(remove_deps)
                  task_to_edit["dependencies"] = list(current_deps_set)
                  
              else:
                  raise ValueError(f"Invalid action '{action}' for task {task_id}")
  
          # --- 循环依赖检测阶段 ---
          temp_tasks_list = list(temp_tasks_map.values())
          for task in temp_tasks_list:
              if self._detect_circular_dependency(task["id"], task["dependencies"], tasks_list=temp_tasks_list):
                  raise ValueError(f"Circular dependency detected for task {task['id']} after applying edits.")
  
          # --- 应用阶段 ---
          self.plan_data["tasks"] = temp_tasks_list
          self._update_timestamp()
          
          results = [{"task_id": edit["task_id"], "new_dependencies": temp_tasks_map[edit["task_id"]]["dependencies"]} for edit in edits]
  
          return {
              "success": True,
              "message": "Tasks dependencies updated successfully.",
              "data": results
          }
      except ValueError as e:
          return {"success": False, "message": str(e), "data": None}