Skip to content

trading_agent.client.mcp_commands

The trading-agent mcp list command: connects to every MCP server once and lists its tools.

It runs on its own, without the gateway, so a new server can be checked before the desks use it.

Typical usage example:

exit_code = asyncio.run(McpCommands(manager).list())

McpCommands 

McpCommands(manager, output=stdout)

Checks the MCP servers for the user.

Creates the commands.

Parameters:

Name Type Description Default
manager McpClientManager

The connections to check.

 required
output TextIO

Where to print.

 stdout
Source code in src/trading_agent/client/mcp_commands.py 
24
25
26
27
28
29
30
31
32
33
34
35
36
def __init__(
    self,
    manager: mcp_client_manager.McpClientManager,
    output: TextIO = sys.stdout,
):
    """Creates the commands.

    Args:
        manager (mcp_client_manager.McpClientManager): The connections to check.
        output (TextIO): Where to print.
    """
    self._manager = manager
    self._output = output

list async 

list()

Connects to every server, waits for each first attempt, and prints the result.

Returns:

Type Description
int

0 when every server is ready, 1 when one failed or did not answer in time.
Source code in src/trading_agent/client/mcp_commands.py 
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 list(self) -> int:
    """Connects to every server, waits for each first attempt, and prints the result.

    Returns:
        int: 0 when every server is ready, 1 when one failed or did not answer in time.
    """
    if not self._manager.connections:
        self._print('No MCP servers are configured.')
        return 0
    task = asyncio.create_task(self._manager.run())
    try:
        waited = 0.0
        while (
            not self._manager.settled() and waited < _SETTLE_TIMEOUT_SECONDS
        ):
            await asyncio.sleep(_POLL_SECONDS)
            waited += _POLL_SECONDS
        for line in self._manager.status_lines():
            self._print(line)
        for tool in self._manager.tools():
            self._print(
                f'  {tool.server}/{tool.name}: {tool.description[:100]}'
            )
    finally:
        task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await task
    if len(self._manager.tools()) and self._all_ready():
        return 0
    return 1