Skip to content

UseSubAgentTool

UseSubAgentTool — dispatches a task to a named sub-agent.

UseSubAgentTool

Bases: AsyncBaseTool

Async tool that dispatches a task to a registered sub-agent.

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

All sub-agent 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 sub-agents, 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
169
170
171
172
173
174
175
176
class UseSubAgentTool(AsyncBaseTool):
    """Async tool that dispatches a task to a registered sub-agent.

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

    All sub-agent 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 sub-agents,
            keyed by name.
    """

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

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

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

    @property
    def description(self) -> str:
        """Description of the sub-agent dispatch tool."""
        return (
            "Dispatch a task to a named sub-agent and return its result. "
            "Only call this tool with a sub-agent 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 sub-agent
        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 sub-agent to dispatch the task to."
                    ),
                },
                "task": {
                    "type": "string",
                    "description": "The task instruction for the sub-agent.",
                },
            },
            "required": ["name", "task"],
        }

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

        Sets ``current_subagent_name`` before calling ``spec.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 sub-agent's ``result.content`` on success, or
                an error result if the sub-agent raises for any reason.
        """
        subagent_name = tool_call.arguments.get("name")
        task_instruction = tool_call.arguments.get("task")

        if not isinstance(subagent_name, str):
            return ToolCallResult(
                tool_call_id=tool_call.id_,
                error=True,
                content=json.dumps(
                    {
                        "error_type": "ValueError",
                        "message": "'name' argument must be a string.",
                    },
                ),
            )
        if not isinstance(task_instruction, str):
            return ToolCallResult(
                tool_call_id=tool_call.id_,
                error=True,
                content=json.dumps(
                    {
                        "error_type": "ValueError",
                        "message": "'task' argument must be a string.",
                    },
                ),
            )
        spec = self._subagents_registry.get(subagent_name)
        token = current_subagent_name.set(subagent_name)
        try:
            if spec is None:
                raise SubAgentNotFoundError(
                    f"Sub-agent '{subagent_name}' not found.",
                )
            result = await spec.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 sub-agent dispatch tool.

description property 

description

Description of the sub-agent dispatch tool.

parameters_json_schema property 

parameters_json_schema

JSON schema for tool parameters.

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

__init__ 

__init__(subagents_registry)

Initialise with a registry of sub-agents.

Parameters:

Name Type Description Default
subagents_registry dict[str, SubAgentSpec]

Sub-agents 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:
    """Initialise with a registry of sub-agents.

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

__call__ async 

__call__(tool_call, *args, **kwargs)

Dispatch a task to the named sub-agent and return its result.

Sets current_subagent_name before calling spec.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 sub-agent's result.content on success, or an error result if the sub-agent 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
169
170
171
172
173
174
175
176
async def __call__(
    self,
    tool_call: ToolCall,
    *args: Any,
    **kwargs: Any,
) -> ToolCallResult:
    """Dispatch a task to the named sub-agent and return its result.

    Sets ``current_subagent_name`` before calling ``spec.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 sub-agent's ``result.content`` on success, or
            an error result if the sub-agent raises for any reason.
    """
    subagent_name = tool_call.arguments.get("name")
    task_instruction = tool_call.arguments.get("task")

    if not isinstance(subagent_name, str):
        return ToolCallResult(
            tool_call_id=tool_call.id_,
            error=True,
            content=json.dumps(
                {
                    "error_type": "ValueError",
                    "message": "'name' argument must be a string.",
                },
            ),
        )
    if not isinstance(task_instruction, str):
        return ToolCallResult(
            tool_call_id=tool_call.id_,
            error=True,
            content=json.dumps(
                {
                    "error_type": "ValueError",
                    "message": "'task' argument must be a string.",
                },
            ),
        )
    spec = self._subagents_registry.get(subagent_name)
    token = current_subagent_name.set(subagent_name)
    try:
        if spec is None:
            raise SubAgentNotFoundError(
                f"Sub-agent '{subagent_name}' not found.",
            )
        result = await spec.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,
    )