Coverage for merco/tools/task_tools.py: 83%

1"""任务委派工具 - 子代理调度""" 

2 

3from .base import BaseTool 

4 

5 

6class TaskTool(BaseTool): 

7 """委派任务给子代理""" 

8 

9 name = "task" 

10 description = "创建任务并派发给子代理执行" 

11 toolset = "task" 

12 parameters = { 

13 "type": "object", 

14 "properties": { 

15 "title": {"type": "string", "description": "任务标题"}, 

16 "description": {"type": "string", "description": "详细描述"}, 

17 "priority": {"type": "integer", "description": "优先级 0=低 1=中 2=高", "default": 1}, 

18 "agent": {"type": "string", "description": "指定子代理名称", "default": "default"}, 

19 }, 

20 "required": ["title"], 

21 } 

22 

23 def __init__(self): 

24 super().__init__() 

25 self._todo_manager = None 

26 self._sub_agent_manager = None 

27 

28 def check(self) -> bool: 

29 """激活！""" 

30 return True 

31 

32 async def execute(self, title: str, description: str = "", priority: int = 1, agent: str = "default") -> dict: 

33 # 1. 创建 Todo 

34 todo = self._todo_manager.create(title, description, priority) 

35 

36 # 2. 派发子代理 

37 subagent_id = await self._sub_agent_manager.dispatch(todo.id, description, agent) 

38 

39 return { 

40 "todo_id": todo.id, 

41 "subagent_id": subagent_id, 

42 "status": "dispatched", 

43 } 

44 

45 

46from .registry import tool_registry # noqa: E402 — 模块末尾自注册 

47tool_registry.register(TaskTool())