Skip to content

trading_agent.tools.delegate_task_tool

The delegate_task tool: hands one to four self-contained tasks to subagents, which run side by side.

Typical usage example:

tool = DelegateTaskTool(subagent_runner)

DelegatedTask

Bases: BaseModel

One task for one subagent.

Attributes:

Name Type Description
subagent str

The subagent's name, from the Subagents section of your instructions.
task str

What it should do, with every fact it needs; it cannot see your conversation.

DelegateTaskArguments

Bases: BaseModel

The arguments of delegate_task.

Attributes:

Name Type Description
tasks list[DelegatedTask]

One to four tasks, run at the same time.

DelegateTaskTool 

DelegateTaskTool(runner)

Bases: Tool

Runs subagent tasks and returns their reports.

Creates the tool.

Parameters:

Name Type Description Default
runner SubagentRunner

The desk's subagent runner.

 required
Source code in src/trading_agent/tools/delegate_task_tool.py 
50
51
52
53
54
55
56
def __init__(self, runner: subagent_runner_module.SubagentRunner):
    """Creates the tool.

    Args:
        runner (subagent_runner_module.SubagentRunner): The desk's subagent runner.
    """
    self._runner = runner

run async 

run(arguments)

Runs the tasks and joins their reports.

Parameters:

Name Type Description Default
arguments DelegateTaskArguments

The tasks.

 required

Returns:

Type Description
ToolResult

One block per task; an error only when no task produced a report.
Source code in src/trading_agent/tools/delegate_task_tool.py 
 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
async def run(self, arguments: DelegateTaskArguments) -> tool.ToolResult:
    """Runs the tasks and joins their reports.

    Args:
        arguments (DelegateTaskArguments): The tasks.

    Returns:
        tool.ToolResult: One block per task; an error only when no task produced a report.
    """
    tasks = []
    for delegated in arguments.tasks:
        tasks.append(
            subagent_runner_module.SubagentTask(
                subagent=delegated.subagent,
                task=delegated.task,
            )
        )
    reports = await self._runner.run_all(tasks)
    blocks = []
    answered = 0
    for report in reports:
        heading = f'### {report.subagent}: {report.outcome}, {report.seconds:.0f} s'
        if report.backend_name:
            heading = f'{heading}, answered by {report.backend_name}'
        lines = [
            heading,
        ]
        for note in report.notes:
            lines.append(f'(note: {note})')
        if report.outcome == 'answered':
            answered += 1
            lines.append(
                untrusted_text.UntrustedText().wrap(
                    f'the subagent {report.subagent}, which may have read outside sources',
                    report.text,
                )
            )
        else:
            lines.append(report.text)
        blocks.append('\n'.join(lines))
    text = '\n\n'.join(blocks)
    if answered == 0:
        return tool.ToolResult.error(text)
    return tool.ToolResult.success(text)