Skip to content

trading_agent.backends.claude_tool_bridge

Hands a desk's tools to Claude Code as an in-process MCP server.

Claude Code calls each tool through the MCP protocol, but the call never leaves this process: the handler runs the call through the desk's ToolCallExecutor, so Claude's calls get the same loop guard, argument checks, time limit and output cap as Gemma's.

Typical usage example:

bridge = ClaudeToolBridge(executor) server = bridge.build_server() bridge.current_trigger = trigger

SERVER_NAME module-attribute 

SERVER_NAME = 'trading_agent'

ClaudeToolHandler 

ClaudeToolHandler(bridge, tool_name)

The handler Claude Code calls for one tool.

Creates the handler.

Parameters:

Name Type Description Default
bridge ClaudeToolBridge

The bridge that knows the current trigger and the executor.

 required
tool_name str

The tool's name in the registry.

 required
Source code in src/trading_agent/backends/claude_tool_bridge.py 
26
27
28
29
30
31
32
33
34
def __init__(self, bridge: ClaudeToolBridge, tool_name: str):
    """Creates the handler.

    Args:
        bridge (ClaudeToolBridge): The bridge that knows the current trigger and the executor.
        tool_name (str): The tool's name in the registry.
    """
    self._bridge = bridge
    self._tool_name = tool_name

__call__ async 

__call__(arguments)

Runs the tool call and returns the result in MCP's shape.

Parameters:

Name Type Description Default
arguments dict[str, Any]

The arguments Claude sent.

 required

Returns:

Type Description
dict[str, Any]

{"content": [{"type": "text", "text": ...}], "is_error": ...}.
Source code in src/trading_agent/backends/claude_tool_bridge.py 
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
async def __call__(self, arguments: dict[str, Any]) -> dict[str, Any]:
    """Runs the tool call and returns the result in MCP's shape.

    Args:
        arguments (dict[str, Any]): The arguments Claude sent.

    Returns:
        dict[str, Any]: {"content": [{"type": "text", "text": ...}], "is_error": ...}.
    """
    call = messages.ToolCall.new(self._tool_name, dict(arguments))
    result = await self._bridge.executor.execute(
        self._bridge.current_trigger,
        call,
    )
    content = [
        {
            'type': 'text',
            'text': result.text,
        },
    ]
    if result.image is not None:
        content.append(
            {
                'type': 'image',
                'data': result.image.base64_data(),
                'mimeType': result.image.media_type,
            }
        )
    return {
        'content': content,
        'is_error': result.is_error,
    }

ClaudeToolBridge 

ClaudeToolBridge(executor)

Builds the MCP server and remembers which trigger the current turn belongs to.

Attributes:

Name Type Description
executor

What runs each call.
current_trigger Trigger | None

The trigger of the turn in progress, set by the turn runner before each turn.

Creates the bridge.

Parameters:

Name Type Description Default
executor ToolCallExecutor

What runs each call.

 required
Source code in src/trading_agent/backends/claude_tool_bridge.py 
78
79
80
81
82
83
84
85
def __init__(self, executor: tool_call_executor_module.ToolCallExecutor):
    """Creates the bridge.

    Args:
        executor (tool_call_executor_module.ToolCallExecutor): What runs each call.
    """
    self.executor = executor
    self.current_trigger: trigger_module.Trigger | None = None

build_server 

build_server()

Builds the in-process MCP server holding every tool in the registry.

Returns:

Type Description
McpSdkServerConfig

The server, for ClaudeAgentOptions.mcp_servers.
Source code in src/trading_agent/backends/claude_tool_bridge.py 
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def build_server(self) -> claude_agent_sdk.McpSdkServerConfig:
    """Builds the in-process MCP server holding every tool in the registry.

    Returns:
        claude_agent_sdk.McpSdkServerConfig: The server, for ClaudeAgentOptions.mcp_servers.
    """
    sdk_tools = []
    for definition in self.executor.tool_registry.definitions():
        function = definition['function']
        sdk_tools.append(
            claude_agent_sdk.SdkMcpTool(
                name=function['name'],
                description=function['description'],
                input_schema=function['parameters'],
                handler=ClaudeToolHandler(self, function['name']),
            )
        )
    return claude_agent_sdk.create_sdk_mcp_server(
        SERVER_NAME,
        tools=sdk_tools,
    )

claude_tool_names 

claude_tool_names()

The names Claude Code knows the tools by.

Returns:

Type Description
list[str]

Each tool's name as "mcp__trading_agent__".
Source code in src/trading_agent/backends/claude_tool_bridge.py 
109
110
111
112
113
114
115
116
117
118
def claude_tool_names(self) -> list[str]:
    """The names Claude Code knows the tools by.

    Returns:
        list[str]: Each tool's name as "mcp__trading_agent__<name>".
    """
    names = []
    for name in self.executor.tool_registry.tool_names:
        names.append(f'mcp__{SERVER_NAME}__{name}')
    return names