Spaces:
Paused
Paused
File size: 4,627 Bytes
9792ea7 | 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 133 | # -*- coding: utf-8 -*-
"""The creating task tool class."""
from typing import Any
from pydantic import BaseModel, Field
from ._task_tool_base import _TaskToolBase
from .._response import ToolChunk
from ...state import AgentState, Task
from ...exception import DeveloperOrientedException
from ...message import TextBlock, ToolResultState
class _TaskCreateParams(BaseModel):
"""The params of the creating task tool."""
subject: str = Field(description="A brief title for the task")
description: str = Field(description="What needs to be done")
metadata: dict[str, Any] | None = Field(
default=None,
description="Arbitrary metadata to attach to the task",
)
class TaskCreate(_TaskToolBase):
"""Create a task for the agent to perform."""
name: str = "TaskCreate"
description: str = """Use this tool to create a structured task list for \
your current session. This helps you track progress, organize complex tasks, \
and demonstrate thoroughness to the user.
It also helps the user understand the progress of the task and overall \
progress of their requests.
## When to Use This Tool
Use this tool proactively in these scenarios:
- Complex multi-step tasks - When a task requires 3 or more distinct steps \
or actions
- Non-trivial and complex tasks - Tasks that require careful planning or \
multiple operations
- Plan mode - When using plan mode, create a task list to track the work
- User explicitly requests todo list - When the user directly asks you to \
use the todo list
- User provides multiple tasks - When users provide a list of things to be \
done (numbered or comma-separated)
- After receiving new instructions - Immediately capture user requirements \
as tasks
- When you start working on a task - Mark it as in_progress BEFORE \
beginning work
- After completing a task - Mark it as completed and add any new follow-up \
tasks discovered during implementation
## When NOT to Use This Tool
Skip using this tool when:
- There is only a single, straightforward task
- The task is trivial and tracking it provides no organizational benefit
- The task can be completed in less than 3 trivial steps
- The task is purely conversational or informational
NOTE that you should not use this tool if there is only one trivial task to \
do. In this case you are better off just doing the task directly.
## Task Fields
- **subject**: A brief, actionable title in imperative form (e.g., \
"Fix authentication bug in login flow")
- **description**: What needs to be done
All tasks are created with status `pending`.
## Tips
- Create tasks with clear, specific subjects that describe the outcome
- After creating tasks, use TaskUpdate to set up dependencies \
(blocks/blockedBy) if needed
- Check TaskList first to avoid creating duplicate tasks"""
input_schema: dict = _TaskCreateParams.model_json_schema()
async def call(
self,
_agent_state: AgentState,
subject: str,
description: str,
metadata: dict[str, Any] | None = None,
) -> ToolChunk:
"""Create the subtask and add it into the agent state."""
if not isinstance(_agent_state, AgentState):
# Expose error to the developer
raise DeveloperOrientedException(
f"Error: TaskCreate requires AgentState to be provided, got "
f"{_agent_state} instead.",
)
try:
# Derive the next sequential id from existing tasks.
# Existing ids that look numeric are considered; any
# non-numeric ids (e.g. legacy UUIDs) are ignored.
max_numeric = 0
for t in _agent_state.tasks_context.tasks:
try:
max_numeric = max(max_numeric, int(t.id))
except (ValueError, TypeError):
pass
next_id = str(max_numeric + 1)
task = Task(
id=next_id,
subject=subject,
description=description,
metadata=metadata or {},
)
_agent_state.tasks_context.tasks.append(task)
return ToolChunk(
content=[
TextBlock(
text=f"Task (id={next_id}) created successfully: "
f"{task.subject}",
),
],
)
except Exception as e:
return ToolChunk(
content=[
TextBlock(text=f"CreateTaskError: {e}"),
],
state=ToolResultState.ERROR,
)
|