trading_agent.tools.tool_call_executor

Runs one tool call the same way for every backend.

Whether the call comes from the agent loop (Gemma, Qwen, OpenAI) or from Claude Code, it is published, checked by the loop guard and the permission gate, run through the registry, and its result published. Keeping that in one place means every backend gets the same protections.

Typical usage example:

executor = ToolCallExecutor('main', registry, event_bus, loop_guard) executor.start_turn() result = await executor.execute(trigger, call)

CompletedCall dataclass

CompletedCall(name, arguments, result_text, is_error)

A tool call that ran, or was held for permission, during a turn.

Attributes:

Name	Type	Description
name	str	The tool.
arguments	dict[str, Any]	The arguments it was called with.
result_text	str	What it returned.
is_error	bool	Whether it failed.

ToolCallExecutor

ToolCallExecutor(desk_name, tool_registry, event_bus, loop_guard, permission_gate=None)

Runs a desk's tool calls through the loop guard and the registry.

Attributes:

Name	Type	Description
tool_registry		The desk's tools.

Creates the executor.

Parameters:

Name	Type	Description	Default
desk_name	str	The desk the calls belong to, used to tag events.	required
tool_registry	ToolRegistry	The desk's tools.	required
event_bus	EventBus	Where calls and results are published.	required
loop_guard	LoopGuard	The guard against repeated identical calls.	required
permission_gate	PermissionGate | None	Holds calls that need the user's permission, or None when none do.	None

Source code in src/trading_agent/tools/tool_call_executor.py

def __init__(
    self,
    desk_name: str,
    tool_registry: tool_registry_module.ToolRegistry,
    event_bus: event_bus_module.EventBus,
    loop_guard: loop_guard_module.LoopGuard,
    permission_gate: permission_gate_module.PermissionGate | None = None,
):
    """Creates the executor.

    Args:
        desk_name (str): The desk the calls belong to, used to tag events.
        tool_registry (tool_registry_module.ToolRegistry): The desk's tools.
        event_bus (event_bus_module.EventBus): Where calls and results are published.
        loop_guard (loop_guard_module.LoopGuard): The guard against repeated identical calls.
        permission_gate (permission_gate_module.PermissionGate | None): Holds calls that need the user's permission, or None when none do.
    """
    self._desk_name = desk_name
    self._permission_gate = permission_gate
    self._completed: dict[str, list[CompletedCall]] = {}
    self.tool_registry = tool_registry
    self._event_bus = event_bus
    self._loop_guard = loop_guard

start_turn

start_turn()

Resets the loop guard at the start of a turn.

Source code in src/trading_agent/tools/tool_call_executor.py

def start_turn(self) -> None:
    """Resets the loop guard at the start of a turn."""
    self._loop_guard.start_turn()

execute async

execute(trigger, call)

Publishes a call, runs it unless the guard refuses it, and publishes the result.

Parameters:

Name	Type	Description	Default
trigger	Trigger	The trigger of the current turn.	required
call	ToolCall	The call to run.	required

Returns:

Type	Description
ToolResult	The tool's result, the guard's refusal, or a note that the call waits for the user's permission.

Source code in src/trading_agent/tools/tool_call_executor.py

async def execute(
    self,
    trigger: trigger_module.Trigger,
    call: messages.ToolCall,
) -> tool_module.ToolResult:
    """Publishes a call, runs it unless the guard refuses it, and publishes the result.

    Args:
        trigger (trigger_module.Trigger): The trigger of the current turn.
        call (messages.ToolCall): The call to run.

    Returns:
        tool_module.ToolResult: The tool's result, the guard's refusal, or a note that the call waits for the user's permission.
    """
    self._publish(
        event_module.EventKind.TOOL_CALL,
        {
            'trigger_id': trigger.trigger_id,
            'call_id': call.call_id,
            'name': call.name,
            'arguments': call.arguments,
        },
    )
    gate = self._permission_gate
    if not self._loop_guard.allows(call):
        result = tool_module.ToolResult.error(
            self._loop_guard.refusal_text(call)
        )
    elif gate is not None and gate.needs_permission(self._desk_name, call.name):
        result = gate.hold(self._desk_name, call)
    else:
        result = await self.tool_registry.run(call)
    self._remember(trigger.trigger_id, call, result)
    payload = {
        'trigger_id': trigger.trigger_id,
        'call_id': call.call_id,
        'name': call.name,
        'is_error': result.is_error,
        'text': result.text,
    }
    if result.picture_desk_path is not None:
        payload['picture_path'] = result.picture_desk_path
    self._publish(event_module.EventKind.TOOL_RESULT, payload)
    return result

completed_calls

completed_calls(trigger_id)

The calls that ran, or were held for permission, for one trigger, oldest first.

The model router reads them when a backend fails partway through a turn, so the next backend does not repeat them.

Parameters:

Name	Type	Description	Default
trigger_id	str	The trigger.	required

Returns:

Type	Description
list[CompletedCall]	The calls; empty when none ran or the trigger is too old to be remembered.

Source code in src/trading_agent/tools/tool_call_executor.py

def completed_calls(self, trigger_id: str) -> list[CompletedCall]:
    """The calls that ran, or were held for permission, for one trigger, oldest first.

    The model router reads them when a backend fails partway through a turn, so the next backend does not repeat them.

    Args:
        trigger_id (str): The trigger.

    Returns:
        list[CompletedCall]: The calls; empty when none ran or the trigger is too old to be remembered.
    """
    return list(self._completed.get(trigger_id, []))