IBKR TWS MCP Server

import pytest from unittest.mock import AsyncMock, MagicMock, patch from src.server import ibkr_connect, ibkr_get_positions from src.tws_client import TWSClient # Mock the AppContext for the MCP context @pytest.fixture def mock_app_context(): mock_tws_client = MagicMock(spec=TWSClient) # Mock the connect method to return success mock_tws_client.connect = AsyncMock(return_value=True) # Mock the get_positions method mock_tws_client.get_positions = AsyncMock(return_value=[ {"account": "DU123456", "contract": {"symbol": "VTI"}, "position": 100.0, "avgCost": 200.50} ]) # Mock the full context object structure mock_ctx = MagicMock() mock_ctx.request_context.lifespan_context.tws = mock_tws_client return mock_ctx @pytest.mark.asyncio async def test_ibkr_connect_tool(mock_app_context): """Test the ibkr_connect MCP tool.""" host = "127.0.0.1" port = 7496 clientId = 1 result = await ibkr_connect(mock_app_context, host, port, clientId) assert result['status'] == 'connected' assert result['host'] == host assert result['port'] == port # Verify that the TWS client's connect method was called correctly mock_app_context.request_context.lifespan_context.tws.connect.assert_called_once_with(host, port, clientId) @pytest.mark.asyncio async def test_ibkr_get_positions_tool(mock_app_context): """Test the ibkr_get_positions MCP tool.""" result = await ibkr_get_positions(mock_app_context) # Verify the result structure and content assert isinstance(result, list) assert len(result) == 1 assert result[0]['contract']['symbol'] == 'VTI' # Verify that the TWS client's get_positions method was called mock_app_context.request_context.lifespan_context.tws.get_positions.assert_called_once()