| 1 | """Spawn tool for creating background subagents.""" |
| 2 | |
| 3 | from contextvars import ContextVar |
| 4 | from typing import TYPE_CHECKING, Any |
| 5 | |
| 6 | from nanobot.agent.tools.base import Tool, tool_parameters |
| 7 | from nanobot.agent.tools.schema import StringSchema, tool_parameters_schema |
| 8 | |
| 9 | if TYPE_CHECKING: |
| 10 | from nanobot.agent.subagent import SubagentManager |
| 11 | |
| 12 | |
| 13 | @tool_parameters( |
| 14 | tool_parameters_schema( |
| 15 | task=StringSchema("The task for the subagent to complete"), |
| 16 | label=StringSchema("Optional short label for the task (for display)"), |
| 17 | required=["task"], |
| 18 | ) |
| 19 | ) |
| 20 | class SpawnTool(Tool): |
| 21 | """Tool to spawn a subagent for background task execution.""" |
| 22 | |
| 23 | def __init__(self, manager: "SubagentManager"): |
| 24 | self._manager = manager |
| 25 | self._origin_channel: ContextVar[str] = ContextVar("spawn_origin_channel", default="cli") |
| 26 | self._origin_chat_id: ContextVar[str] = ContextVar("spawn_origin_chat_id", default="direct") |
| 27 | self._session_key: ContextVar[str] = ContextVar("spawn_session_key", default="cli:direct") |
| 28 | |
| 29 | def set_context(self, channel: str, chat_id: str, effective_key: str | None = None) -> None: |
| 30 | """Set the origin context for subagent announcements.""" |
| 31 | self._origin_channel.set(channel) |
| 32 | self._origin_chat_id.set(chat_id) |
| 33 | self._session_key.set(effective_key or f"{channel}:{chat_id}") |
| 34 | |
| 35 | @property |
| 36 | def name(self) -> str: |
| 37 | return "spawn" |
| 38 | |
| 39 | @property |
| 40 | def description(self) -> str: |
| 41 | return ( |
| 42 | "Spawn a subagent to handle a task in the background. " |
| 43 | "Use this for complex or time-consuming tasks that can run independently. " |
| 44 | "The subagent will complete the task and report back when done. " |
| 45 | "For deliverables or existing projects, inspect the workspace first " |
| 46 | "and use a dedicated subdirectory when helpful." |
| 47 | ) |
| 48 | |
| 49 | async def execute(self, task: str, label: str | None = None, **kwargs: Any) -> str: |
| 50 | """Spawn a subagent to execute the given task.""" |
| 51 | return await self._manager.spawn( |
| 52 | task=task, |
| 53 | label=label, |
| 54 | origin_channel=self._origin_channel.get(), |
| 55 | origin_chat_id=self._origin_chat_id.get(), |
| 56 | session_key=self._session_key.get(), |
| 57 | ) |
| 58 |