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.

result_id: str = PydanticUndefined
status: str = PydanticUndefined
result: str | None = None
error: str | None = None
created_at: datetime.datetime = PydanticUndefined
completed_at: datetime.datetime | None = None
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.

agent_id: str = PydanticUndefined

ID of the agent to spawn

task: str = PydanticUndefined

Task description for the spawned agent

context: str | None = None

Additional context to provide to the agent

wait_for_result: bool = True

Whether to wait for the agent to complete

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.

include_disabled: bool = False

Include disabled agents in the list

@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.