Interactive Brokers MCP Server

import math import os from typing import Any, Dict, Optional from dotenv import load_dotenv from mcp.server.fastmcp import FastMCP from .ib import ib_manager def _to_float(value: Any) -> Optional[float]: if value is None: return None try: f = float(value) except Exception: return None if math.isnan(f) or math.isinf(f): return None return f load_dotenv() server = FastMCP( name="ib-mcp", instructions=( "Local MCP server that connects only to IB Gateway to fetch stock and options data." ), ) @server.tool() def get_stock_quote(symbol: str, exchange: str = "SMART", currency: str = "USD") -> Dict[str, Any]: contract = ib_manager.stock(symbol, exchange, currency) data = ib_manager.req_best_effort(contract) return { "symbol": symbol.upper(), "price": _to_float(data.get("last")), "volume": data.get("volume"), "timestamp": data.get("timestamp"), } @server.tool() def get_option_quote( symbol: str, right: str, strike: float, lastTradeDateOrContractMonth: str, exchange: str = "SMART", currency: str = "USD", ) -> Dict[str, Any]: contract = ib_manager.option(symbol, lastTradeDateOrContractMonth, float(strike), right, exchange, currency) data = ib_manager.req_best_effort(contract) bid = _to_float(data.get("bid")) ask = _to_float(data.get("ask")) last = _to_float(data.get("last")) return { "symbol": symbol.upper(), "contract": f"{symbol.upper()} {lastTradeDateOrContractMonth} {strike} {right.upper()}", "bid": bid, "ask": ask, "last": last, "timestamp": data.get("timestamp"), } @server.tool() def list_options( symbol: str, months_ahead: int = 3, right: str = "BOTH", strikes_around: int = 10, include_weeklies: bool = True, exchange: str = "SMART", currency: str = "USD", ) -> Dict[str, Any]: items = ib_manager.list_option_contracts( symbol=symbol, months_ahead=months_ahead, right=right, strikes_around=strikes_around, include_weeklies=include_weeklies, exchange=exchange, currency=currency, ) return {"count": len(items), "items": items} @server.tool() def get_option_quotes_bulk( symbol: str, months_ahead: int = 3, right: str = "BOTH", strikes_around: int = 10, include_weeklies: bool = True, exchange: str = "SMART", currency: str = "USD", max_contracts: int = 200, ) -> Dict[str, Any]: rows = ib_manager.get_option_quotes_bulk( symbol=symbol, months_ahead=months_ahead, right=right, strikes_around=strikes_around, include_weeklies=include_weeklies, exchange=exchange, currency=currency, max_contracts=max_contracts, ) return {"count": len(rows), "rows": rows} @server.tool() def analyze_bull_call_spreads( symbol: str, min_months: int = 4, max_months: int = 12, min_short_cushion_pct: float = 30.0, min_roi_pct: float = 5.0, strikes_around: int = 9999, include_weeklies: bool = False, deep_long_factor: float = 0.8, exchange: str = "SMART", currency: str = "USD", max_contracts: int = 600, ) -> Dict[str, Any]: from datetime import datetime, timezone, timedelta def parse_exp(s: str): try: if len(s) == 8: return datetime.strptime(s, "%Y%m%d").replace(tzinfo=timezone.utc) if len(s) == 6: dt = datetime.strptime(s + "01", "%Y%m%d").replace(tzinfo=timezone.utc) if dt.month == 12: dt2 = dt.replace(year=dt.year + 1, month=1, day=1) else: dt2 = dt.replace(month=dt.month + 1, day=1) return dt2 - timedelta(days=1) except Exception: return None return None # Spot spot_info = ib_manager.req_best_effort(ib_manager.stock(symbol, exchange, currency)) spot = spot_info.get("last") if spot is None: return {"error": f"No spot for {symbol}"} spot_f = float(spot) # Bulk quotes # Use contract-details-based enumeration to avoid invalid combos and speed up spot_info = ib_manager.req_best_effort(ib_manager.stock(symbol, exchange, currency)) spot = spot_info.get("last") spot_f = float(spot) # Expirations window exps_all = ib_manager.get_option_expirations(symbol, exchange) from datetime import datetime, timezone, timedelta now = datetime.now(timezone.utc) min_exp = now + timedelta(days=30 * max(1, min_months)) max_exp_dt = now + timedelta(days=30 * max(1, max_months)) def parse_exp(s: str): try: if len(s) == 8: return datetime.strptime(s, "%Y%m%d").replace(tzinfo=timezone.utc) if len(s) == 6: dt = datetime.strptime(s + "01", "%Y%m%d").replace(tzinfo=timezone.utc) if dt.month == 12: dt2 = dt.replace(year=dt.year + 1, month=1, day=1) else: dt2 = dt.replace(month=dt.month + 1, day=1) return dt2 - timedelta(days=1) except Exception: return None return None exps = [e for e in exps_all if (parse_exp(e) and min_exp <= parse_exp(e) <= max_exp_dt)] # Focus strikes around spot for speed; deep legs need <= deep_long_factor*spot, shorts need >= spot*(1+min_cushion) strike_min = 0.0 strike_max = spot_f * (1 + min_short_cushion_pct / 100.0) * 1.2 details = ib_manager.list_option_details_filtered( symbol=symbol, expirations=exps, right="C", exchange=exchange, strike_min=strike_min, strike_max=strike_max, max_per_expiry=120, ) quotes = ib_manager.get_quotes_for_conids(details) # Filter expirations by months now = datetime.now(timezone.utc) min_exp = now + timedelta(days=30 * max(1, min_months)) max_exp_dt = now + timedelta(days=30 * max(1, max_months)) rows = [r for r in quotes if (parse_exp(r["expiry"]) and min_exp <= parse_exp(r["expiry"]) <= max_exp_dt)] # Prepare map per expiry from collections import defaultdict by_exp: Dict[str, Dict[float, Dict[str, Any]]] = defaultdict(dict) for r in rows: try: by_exp[r["expiry"]][float(r["strike"])] = r except Exception: continue results = [] deep_long_threshold = deep_long_factor * spot_f min_short_cushion = min_short_cushion_pct / 100.0 min_roi = min_roi_pct / 100.0 for expiry, strike_to_quote in by_exp.items(): strikes = sorted(strike_to_quote.keys()) for i, long_strike in enumerate(strikes): if long_strike > deep_long_threshold: continue long_last = strike_to_quote[long_strike].get("last") if long_last is None: continue for short_strike in strikes[i + 1 :]: cushion = (short_strike - spot_f) / spot_f if cushion < min_short_cushion: continue short_last = strike_to_quote[short_strike].get("last") if short_last is None: continue try: net_debit = float(long_last) - float(short_last) except Exception: continue if net_debit <= 0: continue width = short_strike - long_strike max_profit = width - net_debit if max_profit <= 0: continue roi = max_profit / net_debit if roi < min_roi: continue dt_obj = parse_exp(expiry) dte = (dt_obj - now).days if dt_obj else None results.append( { "expiry": expiry, "dte": dte, "long_strike": long_strike, "short_strike": short_strike, "long_last": _to_float(long_last), "short_last": _to_float(short_last), "net_debit": _to_float(net_debit), "width": _to_float(width), "max_profit": _to_float(max_profit), "roi_pct": _to_float(roi * 100), "cushion_pct": _to_float(cushion * 100), } ) results = [r for r in results if r["roi_pct"] is not None] results.sort(key=lambda r: r["roi_pct"], reverse=True) top = results[:20] return { "symbol": symbol.upper(), "spot": _to_float(spot_f), "total": len(results), "top20": top, } @server.tool() def get_option_quotes_for_date( symbol: str, expiry_yyyymmdd: str, right: str = "BOTH", strikes_around: int = 10, exchange: str = "SMART", currency: str = "USD", max_contracts: int = 300, timeout_sec: Optional[float] = 8.0, enable_fallback: bool = False, prefer_valid_contracts: bool = True, ) -> Dict[str, Any]: """Return quotes for all options of a symbol that expire on a specific YYYYMMDD date. By default enumerates only valid contracts via contract details to avoid security-definition errors, then fetches batched snapshots. Historical fallback can fill missing last. """ items = [] contracts = [] snapshots = [] if prefer_valid_contracts: # Enumerate valid contracts for the expiry using contract details exps = [expiry_yyyymmdd] rights = ([right.upper()] if right.upper() in ("C", "P") else ["C", "P"]) details: list[dict] = [] for r in rights: details.extend( ib_manager.list_option_details_filtered( symbol=symbol, expirations=exps, right=r, exchange=exchange, strike_min=None, strike_max=None, max_per_expiry=0, # no cap ) ) if max_contracts and len(details) > max_contracts: details = details[:max_contracts] # Build from details items = [ { "symbol": d["symbol"], "expiry": d["expiry"], "strike": d["strike"], "right": d["right"], "exchange": d.get("exchange", exchange), "currency": d.get("currency", currency), "conId": d.get("conId"), "localSymbol": d.get("localSymbol"), "tradingClass": d.get("tradingClass"), } for d in details ] # Build Option contracts (set conId/localSymbol when present) and request snapshots for it in items: c = ib_manager.option( it["symbol"], it["expiry"], it["strike"], it["right"], it.get("exchange", exchange), it.get("currency", currency) ) try: if it.get("conId"): c.conId = int(it.get("conId")) except Exception: pass if it.get("localSymbol"): c.localSymbol = it.get("localSymbol") if it.get("tradingClass"): c.tradingClass = it.get("tradingClass") contracts.append(c) snapshots = ib_manager.req_snapshots_batch(contracts, timeout_sec=timeout_sec, chunk_size=80) if items else [] else: # Fallback to fast chain enumeration (may contain invalid combos) items = ib_manager.list_option_contracts( symbol=symbol, months_ahead=2, right=right, strikes_around=strikes_around, include_weeklies=True, exchange=exchange, currency=currency, ) items = [it for it in items if it.get("expiry") == expiry_yyyymmdd] if max_contracts and len(items) > max_contracts: items = items[:max_contracts] contracts = [ ib_manager.option( it["symbol"], it["expiry"], it["strike"], it["right"], it.get("exchange", exchange), it.get("currency", currency) ) for it in items ] snapshots = ib_manager.req_snapshots_batch(contracts, timeout_sec=timeout_sec, chunk_size=80) if items else [] rows = [] for it, snap in zip(items, snapshots): rows.append({ "contract": f"{it['symbol']} {it['expiry']} {it['strike']} {it['right']}", "type": ("Call" if it["right"].upper() == "C" else ("Put" if it["right"].upper() == "P" else it["right"].upper())), "bid": _to_float(snap.get("bid")), "ask": _to_float(snap.get("ask")), "last": _to_float(snap.get("last")), }) # Historical fallback for missing last if requested if enable_fallback and items: for idx, row in enumerate(rows): if row.get("last") is not None: continue try: data = ib_manager.req_best_effort(contracts[idx]) except Exception: data = {} if row.get("last") is None and data.get("last") is not None: row["last"] = _to_float(data.get("last")) # Sort by strike then type for readability def _parse_strike(contract_str: str) -> float: try: return float(contract_str.split()[2]) except Exception: return 0.0 rows.sort(key=lambda r: (_parse_strike(r["contract"]), r["type"])) return { "symbol": symbol.upper(), "expiry": expiry_yyyymmdd, "count": len(rows), "rows": rows, } @server.tool() def get_option_quotes_for_date_table( symbol: str, expiry_yyyymmdd: str, right: str = "BOTH", strikes_around: int = 10, exchange: str = "SMART", currency: str = "USD", max_contracts: int = 300, timeout_sec: Optional[float] = 8.0, enable_fallback: bool = False, ) -> str: """Return a markdown table: Contract | Type(Call-Put) | Bid | Ask | Last""" data = get_option_quotes_for_date( symbol=symbol, expiry_yyyymmdd=expiry_yyyymmdd, right=right, strikes_around=strikes_around, exchange=exchange, currency=currency, max_contracts=max_contracts, timeout_sec=timeout_sec, enable_fallback=enable_fallback, ) rows = data.get("rows", []) lines = [ "| Contract | Type(Call-Put) | Bid | Ask | Last |", "|---|---|---:|---:|---:|", ] for r in rows: def f(v: Any) -> str: if v is None: return "" try: fv = float(v) if fv != fv: return "" return f"{fv:.2f}" except Exception: return str(v) lines.append( f"| {r.get('contract','')} | {r.get('type','')} | {f(r.get('bid'))} | {f(r.get('ask'))} | {f(r.get('last'))} |" ) return "\n".join(lines) if __name__ == "__main__": try: ib_manager.connect() except Exception: pass server.run()