80 lines
3.0 KiB
Python

"""Spawn tool for creating background subagents."""
from typing import Any, TYPE_CHECKING
from nanobot.agent.tools.base import Tool
if TYPE_CHECKING:
from nanobot.agent.subagent import SubagentManager
class SpawnTool(Tool):
"""
Tool to spawn a subagent for background task execution.
The subagent runs asynchronously and announces its result back
to the main agent when complete.
"""
def __init__(self, manager: "SubagentManager"):
self._manager = manager
self._origin_channel = "cli"
self._origin_chat_id = "direct"
def set_context(self, channel: str, chat_id: str) -> None:
"""Set the origin context for subagent announcements."""
self._origin_channel = channel
self._origin_chat_id = chat_id
@property
def name(self) -> str:
return "spawn"
@property
def description(self) -> str:
return (
"Spawn a subagent to handle a task in the background. "
"Use this for complex or time-consuming tasks that can run independently. "
"The subagent will complete the task and report back when done.\n\n"
"CRITICAL: The 'task' parameter MUST be a natural language description of what to do, "
"NOT a tool call. The subagent will figure out how to accomplish the task using its own tools.\n\n"
"CORRECT examples:\n"
"- task='Read all documentation files in the project and create a summary'\n"
"- task='Analyze the codebase structure and generate a report'\n"
"- task='Search for information about X and compile findings'\n\n"
"WRONG (do not use tool call syntax):\n"
"- task='read_dir(path=\"/path/to/file\")'\n"
"- task='read_file(path=\"file.txt\")'"
)
@property
def parameters(self) -> dict[str, Any]:
return {
"type": "object",
"properties": {
"task": {
"type": "string",
"description": (
"A natural language description of the task for the subagent to complete. "
"DO NOT use tool call syntax. Examples: 'Read all documentation and summarize', "
"'Analyze the codebase structure', 'Search the web for X and compile findings'. "
"The subagent will determine which tools to use to accomplish this task."
),
},
"label": {
"type": "string",
"description": "Optional short label for the task (for display)",
},
},
"required": ["task"],
}
async def execute(self, task: str, label: str | None = None, **kwargs: Any) -> str:
"""Spawn a subagent to execute the given task."""
return await self._manager.spawn(
task=task,
label=label,
origin_channel=self._origin_channel,
origin_chat_id=self._origin_chat_id,
)