Spaces:
Running on CPU Upgrade
Running on CPU Upgrade
File size: 4,538 Bytes
258ab19 eaf2575 2523e77 258ab19 eaf2575 258ab19 2523e77 258ab19 0d57ce5 258ab19 0d57ce5 258ab19 0d57ce5 258ab19 2523e77 258ab19 6735687 66647c8 6735687 258ab19 eaf2575 258ab19 eaf2575 258ab19 eaf2575 258ab19 eaf2575 258ab19 eaf2575 258ab19 2523e77 258ab19 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | from typing import Any, Dict, List
from agent.core.session import Event
from agent.utils.terminal_display import format_plan_tool_output
from .types import ToolResult
# In-memory storage for the current plan (raw structure from agent)
_current_plan: List[Dict[str, str]] = []
class PlanTool:
"""Tool for managing a list of todos with status tracking."""
def __init__(self, session: Any = None):
self.session = session
async def execute(self, params: Dict[str, Any]) -> ToolResult:
"""
Execute the WritePlan operation.
Args:
params: Dictionary containing:
- todos: List of todo items, each with id, content, and status
Returns:
ToolResult with formatted output
"""
global _current_plan
todos = params.get("todos", [])
# Validate todos structure
for todo in todos:
if not isinstance(todo, dict):
return {
"formatted": "Error: Each todo must be an object. Re call the tool with correct format (mandatory).",
"isError": True,
}
required_fields = ["id", "content", "status"]
for field in required_fields:
if field not in todo:
return {
"formatted": f"Error: Todo missing required field '{field}'. Re call the tool with correct format (mandatory).",
"isError": True,
}
# Validate status
valid_statuses = ["pending", "in_progress", "completed"]
if todo["status"] not in valid_statuses:
return {
"formatted": f"Error: Invalid status '{todo['status']}'. Must be one of: {', '.join(valid_statuses)}. Re call the tool with correct format (mandatory).",
"isError": True,
}
# Store the raw todos structure in memory
_current_plan = todos
# Emit plan update event if session is available
if self.session:
await self.session.send_event(
Event(
event_type="plan_update",
data={"plan": todos},
)
)
# Format only for display using terminal_display utility
formatted_output = format_plan_tool_output(todos)
return {
"formatted": formatted_output,
"totalResults": len(todos),
"isError": False,
}
def get_current_plan() -> List[Dict[str, str]]:
"""Get the current plan (raw structure)."""
return _current_plan
# Tool specification
PLAN_TOOL_SPEC = {
"name": "plan_tool",
"description": (
"Track progress on multi-step tasks with a todo list (pending/in_progress/completed).\n\n"
"Use for tasks with 3+ steps. Each call replaces the entire plan (send full list).\n\n"
"Rules: exactly ONE task in_progress at a time. Mark completed immediately after finishing. "
"Only mark completed when the task fully succeeded — keep in_progress if there are errors. "
"Update frequently so the user sees progress."
),
"parameters": {
"type": "object",
"properties": {
"todos": {
"type": "array",
"description": "List of todo items",
"items": {
"type": "object",
"properties": {
"id": {
"type": "string",
"description": "Unique identifier for the todo",
},
"content": {
"type": "string",
"description": "Description of the todo task",
},
"status": {
"type": "string",
"enum": ["pending", "in_progress", "completed"],
"description": "Current status of the todo",
},
},
"required": ["id", "content", "status"],
},
}
},
"required": ["todos"],
},
}
async def plan_tool_handler(
arguments: Dict[str, Any], session: Any = None
) -> tuple[str, bool]:
tool = PlanTool(session=session)
result = await tool.execute(arguments)
return result["formatted"], not result.get("isError", False)
|