MCPlanManager

Instructions

Implementation Reference

• src/mcplanmanager/app.py:11-24 (handler)

  The main MCP tool handler function, decorated with @mcp.tool() for automatic registration. It validates inputs using Pydantic models (TaskInput) and delegates execution to the PlanManager instance.

  @mcp.tool()
  def initializePlan(goal: str, tasks: List[TaskInput]) -> ToolResponse[dict]:
      """
      初始化或完全替换一个新的任务计划。
  
      Args:
          goal (str): 描述计划总体目标的字符串。
          tasks (List[TaskInput]): 任务对象的列表。每个任务的结构如下:
            - name (str): 任务的名称，在一个计划中应唯一。
            - dependencies (List[Union[str, int]]): 依赖的任务名称或ID列表。
            - reasoning (str): 阐述为何需要此任务。
      """
      task_dicts = [task.model_dump() for task in tasks]
      return plan_manager.initializePlan(goal, task_dicts)

• src/mcplanmanager/models.py:14-22 (schema)

  Pydantic schema/model for the 'tasks' input parameter in initializePlan tool, defining the structure and validation for each task.

  class TaskInput(BaseModel):
      """
      用于初始化计划时，定义单个任务输入的Pydantic模型。
      这为Agent提供了一个清晰、可验证的数据结构。
      """
      name: str
      dependencies: List[Union[str, int]]
      reasoning: str

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

  Generic Pydantic response wrapper used as the return type for initializePlan (ToolResponse[dict]), standardizing tool outputs.

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

• src/mcplanmanager/plan_manager.py:512-553 (helper)

  Core implementation logic in PlanManager class. Handles plan initialization: validates input, assigns IDs, resolves string dependencies to IDs, detects circular dependencies, and sets up the plan data structure.

  def initializePlan(self, goal: str, tasks: List[Dict]) -> Dict:
      """
      初始化计划
      """
      if not tasks:
          return {"success": False, "message": "At least one task is required"}
      
      current_time = datetime.now().isoformat()
      
      # 重置计划数据
      self.plan_data = {
          "meta": {
              "goal": goal,
              "created_at": current_time,
              "updated_at": current_time
          },
          "state": {
              "current_task_id": None,
              "status": "idle"
          },
          "tasks": []
      }
      
      # 处理任务列表
      try:
          # First pass: create tasks and map names to IDs
          processed_tasks, task_name_to_id = self._process_tasks_pass_one(tasks)
          # Second pass: resolve dependencies
          self._process_tasks_pass_two(tasks, processed_tasks, task_name_to_id)
          # Third pass: detect circular dependencies
          self._check_all_circular_dependencies(processed_tasks)
      except ValueError as e:
          return {"success": False, "message": str(e)}
  
      self.plan_data["tasks"] = processed_tasks
      self._update_timestamp()
      
      return {
          "success": True,
          "message": "Plan initialized successfully",
          "data": self.plan_data
      }