Function _agent_call

src/tool_system/tools/agent.py:154–377

(tool_input: dict[str, Any], context: ToolContext)

        return filter_agents_by_mcp_requirements(agents, available_mcp)

    def _agent_call(tool_input: dict[str, Any], context: ToolContext) -> ToolResult:
        prompt = tool_input.get("prompt", "")
        if not prompt:
            raise ToolInputError("prompt is required")

        description = tool_input.get("description", prompt[:50])
        subagent_type = tool_input.get("subagent_type")
        model = tool_input.get("model")
        run_in_background = bool(tool_input.get("run_in_background", False))
        # Chapter-10 / WI-6.1 — optional human-readable name. We
        # validate / register it AFTER agent_id is generated so the
        # collision-on-running check can compare against the registry
        # state under the registry's own atomicity guarantees.
        agent_name = tool_input.get("name")
        if agent_name is not None and not isinstance(agent_name, str):
            raise ToolInputError("name must be a string when provided")
        if isinstance(agent_name, str) and not agent_name.strip():
            agent_name = None  # treat empty/whitespace as absent

        # Resolve agent definition.
        #
        # Routing rules mirror typescript/src/tools/AgentTool/AgentTool.tsx:318-356:
        # - subagent_type provided → use it (explicit wins).
        # - subagent_type omitted, fork gate on → implicit fork via FORK_AGENT.
        # - subagent_type omitted, fork gate off → default to general-purpose.
        agent_definitions = _get_agent_definitions(context)
        is_fork_path = (
            subagent_type is None and is_fork_subagent_enabled(context)
        )

        if is_fork_path:
            # Recursive-fork guard. Primary check: querySource on the parent's
            # options. Secondary check: scan parent messages for the fork
            # boilerplate tag. Either one trips means we are already inside a
            # fork child, so refuse to spawn another.
            parent_query_source = getattr(context.options, "query_source", None)
            if parent_query_source == f"agent:builtin:{FORK_SUBAGENT_TYPE}" or is_in_fork_child(context.messages):
                raise ToolInputError(
                    "Fork is not available inside a forked worker. "
                    "Complete your task directly using your tools."
                )
            agent_def = FORK_AGENT
        elif subagent_type:
            agent_def = find_agent_by_type(agent_definitions, subagent_type)
            if agent_def is None:
                available = [a.agent_type for a in agent_definitions]
                raise ToolInputError(
                    f"Unknown subagent_type: {subagent_type}. "
                    f"Available: {', '.join(available)}"
                )
        else:
            # Default to general-purpose
            agent_def = (
                find_agent_by_type(agent_definitions, "general-purpose")
                or agent_definitions[0]
                if agent_definitions
                else None
            )