Skip to content

UseSubAgentTool

UseSubAgentTool — dispatches a task to a named subagent.

UseSubAgentTool

Bases: AsyncBaseTool

Async tool that dispatches a task to a registered subagent.

Each subagent registered with the parent coordinator is callable via this single tool. The LLM selects a subagent by name (constrained to an enum of registered names) and provides a free-text task instruction. The tool runs the subagent to completion and returns only result.content, keeping trajectories isolated.

All subagent exceptions — including MaxStepsReachedError — are caught and returned as ToolCallResult(error=True) so the coordinator can re-plan rather than crash.

Attributes:

Name Type Description
subagents_registry dict[str, SubAgentSpec]

Registered subagents, keyed by name.
Source code in src/llm_agents_from_scratch/subagents/tools.py 
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
class UseSubAgentTool(AsyncBaseTool):
    """Async tool that dispatches a task to a registered subagent.

    Each subagent registered with the parent coordinator is callable via this
    single tool. The LLM selects a subagent by name (constrained to an enum
    of registered names) and provides a free-text task instruction. The tool
    runs the subagent to completion and returns only ``result.content``,
    keeping trajectories isolated.

    All subagent exceptions — including ``MaxStepsReachedError`` — are caught
    and returned as ``ToolCallResult(error=True)`` so the coordinator can
    re-plan rather than crash.

    Attributes:
        subagents_registry (dict[str, SubAgentSpec]): Registered subagents,
            keyed by name.
    """

    def __init__(self, subagents_registry: dict[str, SubAgentSpec]) -> None:
        """Initialize with a registry of subagents.

        Args:
            subagents_registry (dict[str, SubAgentSpec]): Subagents to
                register, keyed by name.
        """
        self._subagents_registry = subagents_registry

    @property
    def name(self) -> str:
        """Name of the subagent dispatch tool."""
        return "from_scratch__use_subagent"

    @property
    def description(self) -> str:
        """Description of the subagent dispatch tool."""
        return (
            "Dispatch a task to a named subagent and return its result. "
            "Only call this tool with a subagent name from the "
            "<available_subagents> catalog."
        )

    @property
    def parameters_json_schema(self) -> dict[str, Any]:
        """JSON schema for tool parameters.

        The ``name`` field is constrained to an enum of registered subagent
        names so the LLM can only dispatch to agents that exist.
        """
        return {
            "type": "object",
            "properties": {
                "name": {
                    "type": "string",
                    "enum": sorted(self._subagents_registry),
                    "description": (
                        "Name of the subagent to dispatch the task to."
                    ),
                },
                "task": {
                    "type": "string",
                    "description": "The task instruction for the subagent.",
                },
            },
            "required": ["name", "task"],
        }

    async def __call__(
        self,
        tool_call: ToolCall,
        *args: Any,
        **kwargs: Any,
    ) -> ToolCallResult:
        """Dispatch a task to the named subagent and return its result.

        Builds a fresh ``LLMAgent`` from ``spec.builder`` on every call —
        the spec holds a recipe, not a live agent. Whatever the builder
        was given directly (memory stores, MCP providers) is reused
        as-is; only the agent shell is rebuilt.

        Sets ``current_subagent_name`` before calling ``agent.run()`` so
        the ``asyncio.Task`` it creates copies a context with the name
        already set, and resets it once the dispatch settles. This is the
        only call site that ever sets ``current_subagent_name``.

        Safe under normal usage: ``run_step()`` always reaches this method
        through ``asyncio.gather`` (even for a single tool call), so the
        context mutated here is already a fork of the coordinator's own
        context, never the coordinator's context itself — concurrent or
        nested dispatches can't collide, since each fork owns an
        independent copy from the moment it's created, before any
        ``set()`` runs. Calling this tool directly, bypassing
        ``run_step()`` (e.g. for a manual dispatch demo), is the one path
        where ``set()`` lands on whatever context the caller happens to
        be in — still safe as long as nothing else concurrently shares
        that context, which the ``try``/``finally`` reset below
        guarantees for this call in isolation.

        Args:
            tool_call (ToolCall): The tool call to execute.
            *args (Any): Additional positional arguments.
            **kwargs (Any): Additional keyword arguments.

        Returns:
            ToolCallResult: The subagent's ``result.content`` on success, or
                an error result if the subagent raises for any reason.
        """
        if validation_error_details := validate_tool_call_arguments(
            tool_call,
            self.parameters_json_schema,
        ):
            return ToolCallResult(
                tool_call_id=tool_call.id_,
                error=True,
                content=json.dumps(validation_error_details),
            )

        subagent_name: str = tool_call.arguments["name"]
        task_instruction: str = tool_call.arguments["task"]
        # Schema validation above already constrains `name` to this exact
        # registry, so this should never raise KeyError under normal
        # circumstances.
        spec = self._subagents_registry[subagent_name]
        token = current_subagent_name.set(subagent_name)
        try:
            agent = await spec.builder.build()
            result = await agent.run(
                Task(instruction=task_instruction),
                max_steps=spec.max_steps,
                skills_scopes=spec.skills_scopes,
                explicit_only_skills=spec.explicit_only_skills,
            )
        except Exception as e:
            return ToolCallResult(
                tool_call_id=tool_call.id_,
                error=True,
                content=json.dumps(
                    {
                        "error_type": type(e).__name__,
                        "subagent": subagent_name,
                        "message": str(e),
                    },
                ),
            )
        finally:
            current_subagent_name.reset(token)

        return ToolCallResult(
            tool_call_id=tool_call.id_,
            error=False,
            content=result.content,
        )

name property 

name

Name of the subagent dispatch tool.

description property 

description

Description of the subagent dispatch tool.

parameters_json_schema property 

parameters_json_schema

JSON schema for tool parameters.

The name field is constrained to an enum of registered subagent names so the LLM can only dispatch to agents that exist.

__init__ 

__init__(subagents_registry)

Initialize with a registry of subagents.

Parameters:

Name Type Description Default
subagents_registry dict[str, SubAgentSpec]

Subagents to register, keyed by name.

 required
Source code in src/llm_agents_from_scratch/subagents/tools.py 
36
37
38
39
40
41
42
43
def __init__(self, subagents_registry: dict[str, SubAgentSpec]) -> None:
    """Initialize with a registry of subagents.

    Args:
        subagents_registry (dict[str, SubAgentSpec]): Subagents to
            register, keyed by name.
    """
    self._subagents_registry = subagents_registry

__call__ async 

__call__(tool_call, *args, **kwargs)

Dispatch a task to the named subagent and return its result.

Builds a fresh LLMAgent from spec.builder on every call — the spec holds a recipe, not a live agent. Whatever the builder was given directly (memory stores, MCP providers) is reused as-is; only the agent shell is rebuilt.

Sets current_subagent_name before calling agent.run() so the asyncio.Task it creates copies a context with the name already set, and resets it once the dispatch settles. This is the only call site that ever sets current_subagent_name.

Safe under normal usage: run_step() always reaches this method through asyncio.gather (even for a single tool call), so the context mutated here is already a fork of the coordinator's own context, never the coordinator's context itself — concurrent or nested dispatches can't collide, since each fork owns an independent copy from the moment it's created, before any set() runs. Calling this tool directly, bypassing run_step() (e.g. for a manual dispatch demo), is the one path where set() lands on whatever context the caller happens to be in — still safe as long as nothing else concurrently shares that context, which the try/finally reset below guarantees for this call in isolation.

Parameters:

Name Type Description Default
tool_call ToolCall

The tool call to execute.

 required
*args Any

Additional positional arguments.

 ()
**kwargs Any

Additional keyword arguments.

 {}

Returns:

Name Type Description
ToolCallResult ToolCallResult

The subagent's result.content on success, or an error result if the subagent raises for any reason.
Source code in src/llm_agents_from_scratch/subagents/tools.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
async def __call__(
    self,
    tool_call: ToolCall,
    *args: Any,
    **kwargs: Any,
) -> ToolCallResult:
    """Dispatch a task to the named subagent and return its result.

    Builds a fresh ``LLMAgent`` from ``spec.builder`` on every call —
    the spec holds a recipe, not a live agent. Whatever the builder
    was given directly (memory stores, MCP providers) is reused
    as-is; only the agent shell is rebuilt.

    Sets ``current_subagent_name`` before calling ``agent.run()`` so
    the ``asyncio.Task`` it creates copies a context with the name
    already set, and resets it once the dispatch settles. This is the
    only call site that ever sets ``current_subagent_name``.

    Safe under normal usage: ``run_step()`` always reaches this method
    through ``asyncio.gather`` (even for a single tool call), so the
    context mutated here is already a fork of the coordinator's own
    context, never the coordinator's context itself — concurrent or
    nested dispatches can't collide, since each fork owns an
    independent copy from the moment it's created, before any
    ``set()`` runs. Calling this tool directly, bypassing
    ``run_step()`` (e.g. for a manual dispatch demo), is the one path
    where ``set()`` lands on whatever context the caller happens to
    be in — still safe as long as nothing else concurrently shares
    that context, which the ``try``/``finally`` reset below
    guarantees for this call in isolation.

    Args:
        tool_call (ToolCall): The tool call to execute.
        *args (Any): Additional positional arguments.
        **kwargs (Any): Additional keyword arguments.

    Returns:
        ToolCallResult: The subagent's ``result.content`` on success, or
            an error result if the subagent raises for any reason.
    """
    if validation_error_details := validate_tool_call_arguments(
        tool_call,
        self.parameters_json_schema,
    ):
        return ToolCallResult(
            tool_call_id=tool_call.id_,
            error=True,
            content=json.dumps(validation_error_details),
        )

    subagent_name: str = tool_call.arguments["name"]
    task_instruction: str = tool_call.arguments["task"]
    # Schema validation above already constrains `name` to this exact
    # registry, so this should never raise KeyError under normal
    # circumstances.
    spec = self._subagents_registry[subagent_name]
    token = current_subagent_name.set(subagent_name)
    try:
        agent = await spec.builder.build()
        result = await agent.run(
            Task(instruction=task_instruction),
            max_steps=spec.max_steps,
            skills_scopes=spec.skills_scopes,
            explicit_only_skills=spec.explicit_only_skills,
        )
    except Exception as e:
        return ToolCallResult(
            tool_call_id=tool_call.id_,
            error=True,
            content=json.dumps(
                {
                    "error_type": type(e).__name__,
                    "subagent": subagent_name,
                    "message": str(e),
                },
            ),
        )
    finally:
        current_subagent_name.reset(token)

    return ToolCallResult(
        tool_call_id=tool_call.id_,
        error=False,
        content=result.content,
    )