tiphys.tools.spawn

  1"""
  2Spawn agent tool for sub-agent orchestration.
  3
  4Allows agents to delegate tasks to other agents.
  5"""
  6
  7import asyncio
  8from datetime import datetime
  9from uuid import uuid4
 10
 11import structlog
 12from pydantic import BaseModel, Field
 13
 14from tiphys.context import get_context
 15from tiphys.tools.base import tool
 16
 17logger = structlog.get_logger(__name__)
 18
 19
 20class AsyncResult(BaseModel):
 21    """Result of an asynchronous agent execution."""
 22
 23    result_id: str
 24    status: str  # "pending", "completed", "failed"
 25    result: str | None = None
 26    error: str | None = None
 27    created_at: datetime = Field(default_factory=datetime.utcnow)
 28    completed_at: datetime | None = None
 29
 30
 31# Simple in-memory store for async results
 32# In production, this should be part of the AppContext or persistent store
 33_async_results: dict[str, AsyncResult] = {}
 34
 35
 36class SpawnAgentArgs(BaseModel):
 37    """Arguments for spawn_agent tool."""
 38
 39    agent_id: str = Field(..., description="ID of the agent to spawn")
 40    task: str = Field(..., description="Task description for the spawned agent")
 41    context: str | None = Field(
 42        default=None,
 43        description="Additional context to provide to the agent",
 44    )
 45    wait_for_result: bool = Field(
 46        default=True,
 47        description="Whether to wait for the agent to complete",
 48    )
 49
 50
 51class ListAgentsArgs(BaseModel):
 52    """Arguments for list_agents tool."""
 53
 54    include_disabled: bool = Field(
 55        default=False,
 56        description="Include disabled agents in the list",
 57    )
 58
 59
 60@tool(
 61    name="spawn_agent",
 62    description=(
 63        "Delegate a task to another agent. Use this to leverage specialized agents "
 64        "for specific tasks. The spawned agent runs in its own session."
 65    ),
 66)
 67async def spawn_agent(
 68    agent_id: str,
 69    task: str,
 70    context: str | None = None,
 71    wait_for_result: bool = True,
 72) -> str:
 73    """
 74    Spawn a sub-agent to handle a task.
 75
 76    Args:
 77        agent_id: ID of the agent to spawn.
 78        task: Task description for the spawned agent.
 79        context: Additional context to provide.
 80        wait_for_result: Whether to wait for completion.
 81
 82    Returns:
 83        The agent's response, or a result ID if not waiting.
 84    """
 85    # Get AppContext to access registry
 86    try:
 87        app_context = get_context()
 88        agent_registry = app_context.agent_registry
 89
 90        # Build message
 91        message = task
 92        if context:
 93            message = f"{context}\n\n{task}"
 94
 95        # Create session key
 96        session_key = f"spawn:{agent_id}:{uuid4().hex[:8]}"
 97
 98        # Check if agent exists
 99        agent_def = agent_registry.get(agent_id)
100        if not agent_def:
101            # Fallback to creating a transient agent config if not found
102            # This allows spawning "temporary" agents with different models
103            # e.g. spawn_agent("gpt-4o", "analyze this")
104            # For now, just error out
105            available = agent_registry.list_ids()
106            return (
107                f"Agent '{agent_id}' not found. Available agents: {', '.join(available) or 'none'}"
108            )
109
110        if not agent_def.enabled:
111            return f"Agent '{agent_id}' is disabled."
112
113        # Define runner creation logic
114        async def run_sub_agent() -> str:
115            from tiphys.agent.runner import AgentRunner
116            from tiphys.config import TiphysConfig
117
118            # Create sub-config (clone parent but override agent specifics)
119            parent_config = app_context.config
120            sub_config = TiphysConfig(
121                host=parent_config.host,
122                port=parent_config.port,
123                model=agent_def.model or parent_config.model,
124                api_base=agent_def.api_base or parent_config.api_base,
125                api_key=agent_def.api_key or parent_config.api_key,
126                agent_id=agent_def.id,
127                agent_name=agent_def.name,
128                dm_scope=agent_def.dm_scope,
129                data_dir=parent_config.data_dir,
130                plugins_dir=parent_config.plugins_dir,
131                skills_dir=agent_def.skills_dir or parent_config.skills_dir,
132                exec_mode=parent_config.exec_mode,
133                allowed_commands=parent_config.allowed_commands,
134                log_level=parent_config.log_level,
135            )
136
137            # Use AppContext to create runner but inject sub-config
138            # We create a temporary context? Or just pass config override if Runner supported it
139            # Currently Runner takes config in init.
140
141            runner = AgentRunner(
142                config=sub_config,
143                tool_registry=app_context.tool_registry,  # Share tools
144                event_bus=app_context.event_bus,  # Share bus
145            )
146
147            return await runner.run(
148                session_key=session_key,
149                user_message=message,
150                max_steps=agent_def.max_steps,
151            )
152
153        if wait_for_result:
154            try:
155                return await run_sub_agent()
156            except Exception as e:
157                logger.error("Sub-agent failed", agent_id=agent_id, error=str(e))
158                return f"Error running agent '{agent_id}': {e}"
159        else:
160            # Async execution
161            result_id = str(uuid4())
162            _async_results[result_id] = AsyncResult(result_id=result_id, status="pending")
163
164            async def background_run() -> None:
165                try:
166                    result = await run_sub_agent()
167                    _async_results[result_id].status = "completed"
168                    _async_results[result_id].result = result
169                    _async_results[result_id].completed_at = datetime.utcnow()
170                except Exception as e:
171                    _async_results[result_id].status = "failed"
172                    _async_results[result_id].error = str(e)
173                    _async_results[result_id].completed_at = datetime.utcnow()
174
175            asyncio.create_task(background_run())
176            return f"Task started. Result ID: {result_id}. Use get_agent_result('{result_id}') to check status."
177
178    except Exception as e:
179        return f"System error spawning agent: {e}"
180
181
182@tool(
183    name="get_agent_result",
184    description="Get the result of an asynchronous agent task.",
185)
186async def get_agent_result(result_id: str) -> str:
187    """
188    Check the status and get result of an async task.
189    """
190    res = _async_results.get(result_id)
191    if not res:
192        return "Result ID not found."
193
194    if res.status == "pending":
195        return "Task is still running."
196
197    if res.status == "failed":
198        return f"Task failed: {res.error}"
199
200    return f"Task completed:\n\n{res.result}"
201
202
203@tool(
204    name="list_agents",
205    description="List all available agents and their capabilities.",
206)
207async def list_agents(include_disabled: bool = False) -> str:
208    """
209    List available agents.
210    """
211    try:
212        context = get_context()
213        registry = context.agent_registry
214        agents = registry.list_agents()
215
216        if not include_disabled:
217            agents = [a for a in agents if a.enabled]
218
219        if not agents:
220            return "No agents available."
221
222        lines = ["Available agents:", ""]
223        for agent in agents:
224            status = "" if agent.enabled else " (disabled)"
225            lines.append(f"- **{agent.id}**: {agent.name}{status}")
226            if agent.description:
227                lines.append(f"  {agent.description}")
228            if agent.model:
229                lines.append(f"  Model: {agent.model}")
230
231        return "\n".join(lines)
232    except Exception as e:
233        return f"Error listing agents: {e}"