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