IBKR TWS MCP服务器
本项目为交互式经纪人（IBKR）交易者工作站（TWS）API实现了一个模型上下文协议（MCP）服务器。它使用官方 modelcontextprotocol/python-sdk 和 ib_async 将TWS的关键功能作为MCP工具公开，实现与LLM客户的无缝集成，以实现自动化财务工作流程，如投资组合再平衡。
服务器支持 HTTP流 对于MCP协议和 WebSocket流媒体 用于实时数据订阅（市场数据、投资组合更新、新闻简报）。
特性
- MCP合规性： 使用FastMCP构建，以完全遵守具有HTTP流支持的模型上下文协议。
- 异步TWS集成： 杠杆作用
ib_async（ib-insync的维护分支），用于与TWS API进行非阻塞、异步交互。 - 综合工具集： 51+用于连接管理、市场数据检索（历史和流媒体）、账户和投资组合查询以及订单管理（下单和取消订单）的工具。
- MCP提示： 6个指导性工作流程，将多种工具组合成专家级交易策略（支架订单、投资组合再平衡、风险评估、期权策略、市场分析和工作空间设置）。
- 实时WebSocket流媒体： 三个专用的WebSocket端点用于连续实时数据：
- 市场数据流 -多个符号的实时报价 - 投资组合流 -实时投资组合和账户更新 - 新闻流 -IBKR新闻公告和系统消息
- HTTP流传输： 使用分块传输编码实现高效的MCP工具响应。
- CORS支持： 配置为允许来自基于浏览器的MCP客户端（例如Goflow、web应用程序）的跨源请求。
- 容器化部署： 准备好使用Docker和Docker Compose配置。
建筑
┌─────────────────────────────────────────────────┐
│ Client Application │
├─────────────────────────────────────────────────┤
│ 1. MCP Client (HTTP Streaming) │
│ POST /api/v1/mcp │
│ - Tool calls: connect, orders, positions │
│ │
│ 2. WebSocket Clients (real-time streams) │
│ WS /api/v1/stream/market-data │
│ WS /api/v1/stream/portfolio │
│ WS /api/v1/stream/news │
└─────────────────────────────────────────────────┘
│ │
│ HTTP POST │ WebSocket
▼ ▼
┌─────────────────────────────────────────────────┐
│ IBKR TWS MCP Server │
├─────────────────────────────────────────────────┤
│ • FastMCP Server (HTTP streaming) │
│ • WebSocket Handlers (real-time streaming) │
│ • Shared TWS Client (ib_async) │
└─────────────────────────────────────────────────┘
│
│ IB Client Protocol
▼
┌─────────────────────────────────────────────────┐
│ TWS / IB Gateway │
│ (127.0.0.1:7497 or :4001) │
└─────────────────────────────────────────────────┘快速开始
1.安装依赖项
# Using uv (recommended)
uv sync
# Or using pip
pip install -r requirements.txt2.配置环境
# Copy example environment file
cp .env.example .env
# Edit .env with your settings
TWS_HOST=127.0.0.1
TWS_PORT=7497
TWS_CLIENT_ID=1
TWS_PAPER_ACCOUNT=DU2515295
SERVER_HOST=0.0.0.0
SERVER_PORT=80003.CORS配置（浏览器客户端）
服务器配置为允许来自基于浏览器的MCP客户端的跨源请求：
- 允许的来源： 所有起源(
*)-支持外部域，如https://test.lizhao.net - 允许的方法： 获取、发布、放置、删除、选项、头部
- 允许的标头： 所有标头，包括MCP特定标头(
Mcp-Session-Id,Mcp-Initialize-Request) - 资格证书： 已为经过身份验证的请求启用
- 飞行前缓存： 24小时
这使得能够与基于浏览器的MCP客户端（如Goflow）集成，而不受CORS限制。
4.启动TWS/网关
启动Interactive Brokers TWS或IB网关并启用API连接。
5.运行服务器
uv run python main.py5.测试服务器
# Health check
curl http://localhost:8000/health
# MCP tool call (connect to TWS)
curl -X POST http://localhost:8000/api/v1/mcp \
-H "Content-Type: application/json" \
-d '{
"jsonrpc": "2.0",
"method": "tools/call",
"params": {
"name": "ibkr_connect",
"arguments": {"host": "127.0.0.1", "port": 7497, "clientId": 1}
},
"id": 1
}'WebSocket流示例
市场数据流（Python）
import websockets
import json
import asyncio
async def stream_quotes():
async with websockets.connect('ws://localhost:8000/api/v1/stream/market-data') as ws:
# Subscribe to AAPL
await ws.send(json.dumps({
"action": "subscribe",
"symbol": "AAPL",
"secType": "STK",
"exchange": "SMART",
"currency": "USD"
}))
# Receive real-time updates
async for message in ws:
data = json.loads(message)
if data["type"] == "market_data":
print(f"{data['symbol']}: ${data['data']['last']}")
asyncio.run(stream_quotes())市场数据流（JavaScript/浏览器）
const ws = new WebSocket('ws://localhost:8000/api/v1/stream/market-data');
ws.onopen = () => {
ws.send(JSON.stringify({
action: 'subscribe',
symbol: 'AAPL',
secType: 'STK'
}));
};
ws.onmessage = (event) => {
const data = JSON.parse(event.data);
if (data.type === 'market_data') {
console.log(`${data.symbol}: $${data.data.last}`);
}
};有关包括投资组合流和新闻公告在内的完整示例，请参阅 HTTP流示例.
MCP提示
该服务器包括6个全面的提示，为复杂的交易操作提供指导工作流程：
投资组合管理
- setup_trading_workspace -使用流媒体资源完成工作区设置
- 重新平衡投资组合 -通过税务优化和OCA集团重新平衡投资组合
- 评估_投资组合_风险 -采用贝塔加权、VaR和压力测试的风险分析
交易执行
- 执行_订单 -带止盈和止损自动化的支架订单
- 执行选项策略 -期权策略（备兑看涨期权、保护性看跌期权、看跌期权等）
市场分析
- 分析市场条件 -结合技术、基本面、新闻和期权的多维分析
每个提示都提供了分步工作流程，包括工具调用示例、最佳实践、风险警告和示例计算。看 提示指南 了解详情。
入门指南
请参阅 安装指南 有关先决条件、环境配置以及在本地或容器中运行服务器的详细说明。
API 参考
所有公开的MCP工具、其参数和返回类型的完整列表可以在 API 参考.
流媒体文档
端到端测试
该服务器旨在支持投资组合再平衡E2E案例。您可以使用以下工具测试所有功能 Claude MCP检查员.
请参阅 API 参考 有关如何使用检查器与正在运行的服务器交互的分步指南。
项目结构
有关项目组织的详细说明，请参阅 项目_结构.md.
ibkr-tws-mcp-server/
├── src/ # Source code
│ ├── server.py # FastMCP server and tool definitions
│ ├── tws_client.py # TWS client wrapper using ib_async
│ ├── models.py # Pydantic data models
│ └── streaming/ # WebSocket streaming handlers
│ ├── websocket_manager.py
│ ├── market_data.py
│ ├── portfolio.py
│ └── news.py
├── tests/ # Test suite
│ ├── unit/ # Unit tests with mocks
│ └── integration/ # Integration test documentation
├── docs/ # Documentation
│ ├── API.md # MCP tools API reference
│ ├── SETUP.md # Setup and deployment guide
│ ├── HTTP_STREAMING_MIGRATION.md # Streaming migration plan
│ ├── HTTP_STREAMING_EXAMPLES.md # Streaming examples
│ ├── HTTP_STREAMING_SUMMARY.md # Streaming summary
│ └── ... # Additional guides and troubleshooting
├── diagnostics/ # Diagnostic and testing scripts
├── scripts/ # Utility scripts
├── main.py # Application entry point
├── pyproject.toml # Project dependencies
└── README.md # This file