Skip to content

trading_agent.tools.tool_registry

Holds a desk's tools and runs them by name.

The registry is the one place where a tool call from the model meets real code. It checks the arguments, applies a time limit, catches failures and caps the size of the output, so a misbehaving tool cannot crash the desk or flood the model's context.

Typical usage example:

registry = ToolRegistry([GetTimeTool()], timeout_seconds=60) result = await registry.run(tool_call)

ToolRegistry 

ToolRegistry(tools, timeout_seconds)

The tools available to one desk.

Creates a registry.

Parameters:

Name Type Description Default
tools Sequence[Tool]

The tools, each with a unique name.

 required
timeout_seconds float

How long one tool call may run before it is stopped.

 required

Raises:

Type Description
ValueError

Two tools have the same name.
Source code in src/trading_agent/tools/tool_registry.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
def __init__(
    self,
    tools: Sequence[tool_module.Tool],
    timeout_seconds: float,
):
    """Creates a registry.

    Args:
        tools (Sequence[tool_module.Tool]): The tools, each with a unique name.
        timeout_seconds (float): How long one tool call may run before it is stopped.

    Raises:
        ValueError: Two tools have the same name.
    """
    self._tools = {}
    for tool in tools:
        if tool.name in self._tools:
            raise ValueError(f'Duplicate tool name: {tool.name!r}')
        self._tools[tool.name] = tool
    self._timeout_seconds = timeout_seconds

tool_names property 

tool_names

The names of the registered tools, in registration order.

definitions 

definitions()

Describes every tool in the function-calling format Ollama and OpenAI share.

Returns:

Type Description
list[dict[str, Any]]

One {"type": "function", "function": {...}} entry per tool.
Source code in src/trading_agent/tools/tool_registry.py 
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def definitions(self) -> list[dict[str, Any]]:
    """Describes every tool in the function-calling format Ollama and OpenAI share.

    Returns:
        list[dict[str, Any]]: One {"type": "function", "function": {...}} entry per tool.
    """
    definitions = []
    for tool in self._tools.values():
        definitions.append(
            {
                'type': 'function',
                'function': {
                    'name': tool.name,
                    'description': tool.description,
                    'parameters': tool.input_schema(),
                },
            }
        )
    return definitions

run async 

run(call)

Runs one tool call and always returns a result, never raises.

Parameters:

Name Type Description Default
call ToolCall

The call the model made.

 required

Returns:

Type Description
ToolResult

The tool's output, or an error result the model can act on.
Source code in src/trading_agent/tools/tool_registry.py 
 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
async def run(self, call: messages.ToolCall) -> tool_module.ToolResult:
    """Runs one tool call and always returns a result, never raises.

    Args:
        call (messages.ToolCall): The call the model made.

    Returns:
        tool_module.ToolResult: The tool's output, or an error result the model can act on.
    """
    tool = self._tools.get(call.name)
    if tool is None:
        return tool_module.ToolResult.error(
            f'There is no tool named {call.name!r}. Available tools: {", ".join(self._tools)}.'
        )
    try:
        arguments = tool.arguments_model.model_validate(call.arguments)
    except pydantic.ValidationError as error:
        return tool_module.ToolResult.error(
            f'The arguments for {call.name} are invalid: {error}. You sent: {json.dumps(call.arguments)}'
        )
    try:
        result = await asyncio.wait_for(
            tool.run(arguments),
            timeout=self._timeout_for(tool),
        )
    except TimeoutError:
        return tool_module.ToolResult.error(
            f'{call.name} did not finish within {self._timeout_for(tool):g} seconds and was stopped.'
        )
    except Exception as error:
        _LOGGER.exception('Tool %s failed', call.name)
        return tool_module.ToolResult.error(
            f'{call.name} failed: {error!r}'
        )
    return self._capped(result)

close async 

close()

Closes every tool, for example stopping the bash terminal.

Source code in src/trading_agent/tools/tool_registry.py 
111
112
113
114
async def close(self) -> None:
    """Closes every tool, for example stopping the bash terminal."""
    for tool in self._tools.values():
        await tool.close()