"""
spawn_agent — delegates a focused sub-task to an isolated agent instance.
The sub-agent runs its own tool-calling loop with a clean context window.
It cannot spawn further sub-agents (recursion is blocked via exclude_tools).
The result is returned as a plain text summary.
"""
import structlog
from navi.exceptions import ProfileNotFound
from ._internal.base import Tool, ToolContext, ToolResult, current_session_id
log = structlog.get_logger()
class SpawnAgentTool(Tool):
name = "spawn_agent"
description = (
"Delegate EXACTLY ONE step of your plan to an isolated sub-agent with its own "
"context and tool loop.\n\n"
"Use it when a step needs 3+ tool calls as one logical unit — never for a single "
"call, and one call per plan step, never bundling several steps into one "
"sub-agent.\n\n"
"Synchronous by default (blocks until it finishes); background=true detaches and "
"returns a task_id. The user cannot see sub-agent output — synthesise findings "
"into your own response.\n\n"
"Omit profile_id unless the step needs another profile's specialisation.\n\n"
'Details, examples and the profile list: tool_manual("spawn_agent").'
)
parameters = {
"type": "object",
"properties": {
"task": {
"type": "string",
"description": (
"What the sub-agent must accomplish: exact goal, success criteria, "
"expected output format. "
"End with: 'Complete ALL assigned work before responding. Your output is final.'"
),
},
"briefing": {
"type": "string",
"description": (
"Static context injected as system-level instruction: "
"IPs, credentials, file paths, constraints, step-by-step instructions."
),
},
"profile_id": {
"type": "string",
"description": (
"Defaults to the current session's profile — the right choice for most "
"work. Set it only to specialise: 'server_admin' (remote ops), "
"'secretary' (research/writing), 'tool_developer' (Navi tool "
"implementation)."
),
},
"system_prompt": {
"type": "string",
"description": (
"Optional role definition, injected on top of the profile default. "
"e.g. 'You are a security auditor. Report findings by severity.'"
),
},
"max_iterations": {
"type": "integer",
"description": "Maximum tool-call iterations for the sub-agent (default: 40).",
},
"background": {
"type": "boolean",
"description": (
"Detach: returns a task_id immediately; collect via tasks, or wait for the "
"completion note. Use above ~45-60s. Default false (synchronous)."
),
},
"inherit_system_prompt": {
"type": "boolean",
"description": (
"Start from the parent profile's full system prompt as a base, then overlay "
"this sub-agent's own specialisation. Default false — the sub-agent uses only "
"its own subagent prompt."
),
},
},
"required": ["task"],
}
def __init__(
self,
profile_registry,
tool_registry,
backend_registry,
session_store,
memory_store=None,
mcp_manager=None,
) -> None:
self._profile_registry = profile_registry
self._tool_registry = tool_registry
self._backend_registry = backend_registry
self._session_store = session_store
self._memory_store = memory_store
self._mcp_manager = mcp_manager
async def execute(self, params: dict, ctx: ToolContext | None = None) -> ToolResult:
# Import here to avoid module-level circular import
from navi.core.agent import Agent
from navi.tools.scratchpad import get_section
task = params["task"].strip()
briefing = (params.get("briefing") or "").strip() or None
custom_system_prompt = (params.get("system_prompt") or "").strip() or None
max_iterations = int(params.get("max_iterations") or 40)
inherit_system_prompt = bool(params.get("inherit_system_prompt", False))
# task → user message (what to do)
# briefing → system prompt (context, credentials, instructions)
user_message = task
# Resolve profile: explicit override → parent session's profile → first available
# Support both 'profile_id' and 'profile' as parameter names since models sometimes
# use 'profile' when the description says "Profile to use".
profile_id = (params.get("profile_id") or params.get("profile") or "").strip()
if not profile_id:
profile_id = await self._resolve_parent_profile(ctx)
try:
selected_profile = self._profile_registry.get(profile_id)
except ProfileNotFound:
available = ", ".join(p.id for p in self._profile_registry.all())
return ToolResult(
success=False,
output=(
f"Unknown sub-agent profile_id: {profile_id!r}. "
f"Available profiles: {available or '(none)'}."
),
error=f"unknown_profile:{profile_id}",
)
# Read parent scratchpad context_transfer section and pass it to the sub-agent.
parent_sid = ctx.session_id if ctx else current_session_id.get()
context_transfer = (await get_section(parent_sid, "context_transfer")) if parent_sid else ""
scope = selected_profile.get_subagent_tools()
log.info("spawn_agent.start", profile_id=profile_id, max_iterations=max_iterations,
task_preview=task[:80], has_briefing=bool(briefing),
has_context_transfer=bool(context_transfer),
has_system_prompt=bool(custom_system_prompt),
subagent_tools=len(scope.native) + sum(len(v) for v in scope.mcp.values()))
agent = Agent(
session_store=None, # ephemeral — no DB access
profile_registry=self._profile_registry,
tool_registry=self._tool_registry,
backend_registry=self._backend_registry,
workers=[], # no post-response workers for sub-agents
memory_store=self._memory_store,
mcp_manager=self._mcp_manager,
)
try:
result_text, completed = await agent.run_ephemeral(
user_message=user_message,
profile_id=profile_id,
max_iterations=max_iterations,
exclude_tools=["spawn_agent"], # prevent recursion
briefing=briefing,
custom_system_prompt=custom_system_prompt,
inherit_system_prompt=inherit_system_prompt,
context_transfer=context_transfer or None,
parent_session_id=parent_sid,
timeout_seconds=300.0,
)
log.info("spawn_agent.done", profile_id=profile_id, completed=completed,
result_len=len(result_text))
if completed:
output = (
"[Sub-agent completed — USER CANNOT SEE THIS. "
"Synthesise the findings into your own response.]\n\n"
+ result_text
)
else:
output = (
"[Sub-agent hit iteration limit — result may be incomplete. "
"USER CANNOT SEE THIS. "
"Synthesise what was found and note what is missing.]\n\n"
+ result_text
)
return ToolResult(
success=completed,
output=output,
metadata={"completed": completed, "result_len": len(result_text)},
)
except Exception as e:
log.error("spawn_agent.error", error=str(e), exc_info=True)
return ToolResult(success=False, output=f"Sub-agent failed: {e}", error=str(e))
async def _resolve_parent_profile(self, ctx: ToolContext | None = None) -> str:
"""Return the profile of the current parent session, or fallback to first profile."""
session_id = ctx.session_id if ctx else current_session_id.get()
if session_id and self._session_store:
try:
session = await self._session_store.get(session_id)
if session:
return session.profile_id
except Exception:
pass
# Fallback: first registered profile
all_profiles = self._profile_registry.all()
return all_profiles[0].id if all_profiles else "secretary"
