MCPlanManager

Instructions

Implementation Reference

• src/mcplanmanager/app.py:119-133 (handler)

  The primary handler for the MCP 'addTask' tool. Decorated with @mcp.tool() for automatic registration and schema generation from signature/docstring. Delegates execution to PlanManager instance.

  @mcp.tool()
  def addTask(name: str, dependencies: List[int], reasoning: str, after_task_id: Optional[int] = None) -> ToolResponse[TaskOutput]:
      """
      向当前计划中动态添加一个新任务。
  
      Args:
          name (str): 新任务的名称，应确保唯一性。
          dependencies (List[int]): 新任务所依赖的任务ID的整数列表 (从0开始)。
          reasoning (str): 解释为何要添加此任务的字符串。
          after_task_id (int, optional): 一个任务ID，新任务将被插入到该任务之后。如果省略，则添加到列表末尾。
          
      Returns:
          ToolResponse[TaskOutput]: 包含新创建任务的响应对象。
      """
      return plan_manager.addTask(name, dependencies, reasoning, after_task_id)

• src/mcplanmanager/plan_manager.py:212-252 (helper)

  Core business logic implementation for adding a new task to the plan. Performs dependency validation, circular dependency detection, task creation, insertion at specified position or append, and state updates.

  def addTask(self, name: str, dependencies: List[int], reasoning: str, 
              after_task_id: Optional[int] = None) -> Dict:
      """添加新任务到计划中"""
      # 验证依赖任务存在
      for dep_id in dependencies:
          if not self._find_task_by_id(dep_id):
              return {"success": False, "message": f"Dependency task {dep_id} not found"}
      
      new_id = self._get_next_task_id()
      
      # 检测循环依赖
      if self._detect_circular_dependency(new_id, dependencies):
          return {"success": False, "message": "Circular dependency detected"}
          
      new_task = {
          "id": new_id,
          "name": name,
          "status": "pending",
          "dependencies": dependencies,
          "reasoning": reasoning,
          "result": None
      }
      
      # 插入任务
      if after_task_id is not None:
          try:
              # 寻找插入位置
              insert_index = next(i for i, task in enumerate(self.plan_data["tasks"]) if task["id"] == after_task_id) + 1
              self.plan_data["tasks"].insert(insert_index, new_task)
          except StopIteration:
              return {"success": False, "message": f"Task with id {after_task_id} not found"}
      else:
          self.plan_data["tasks"].append(new_task)
          
      self._update_timestamp()
      
      return {
          "success": True,
          "data": new_task,
          "message": "Task added successfully"
      }

• src/mcplanmanager/models.py:33-43 (schema)

  Pydantic model defining the structure of a TaskOutput, which is the 'data' field in the ToolResponse returned by addTask.

  class TaskOutput(BaseModel):
      """
      用于工具函数返回任务信息时，定义单个任务输出的Pydantic模型。
      """
      id: int
      name: str
      status: str
      dependencies: List[int]
      reasoning: str
      result: Optional[str] = None

• src/mcplanmanager/models.py:6-12 (schema)

  Generic Pydantic response wrapper used by addTask and all other tools: ToolResponse[TaskOutput]

  class ToolResponse(BaseModel, Generic[T]):
      """
      一个通用的工具响应模型，用于标准化所有工具的返回结构。
      """
      success: bool = Field(True, description="操作是否成功。")
      message: Optional[str] = Field(None, description="关于操作结果的可读消息。")
      data: Optional[T] = Field(None, description="操作返回的主要数据负载。")