#!/usr/bin/env python3 """ MT5 MCP Server for Linux/Devuan ================================ FastAPI-based HTTP server that provides MCP-compatible endpoints for MetaTrader 5 integration on Linux via Wine + mt5linux. Modes: --mock : Run with simulated data (for testing) --live : Connect to real MT5 via mt5linux rpyc bridge Usage: python mcp_server.py --mock # Test mode python mcp_server.py --live # Real MT5 connection python mcp_server.py --mock --port 8000 """ import os import sys import json import logging import argparse import random import asyncio from datetime import datetime, timedelta, timezone from typing import Dict, Any, Optional, List from contextlib import asynccontextmanager from fastapi import FastAPI, HTTPException, Query from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from pydantic import BaseModel import uvicorn # --------------------------------------------------------------------------- # Logging # --------------------------------------------------------------------------- logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", handlers=[ logging.StreamHandler(sys.stdout), logging.FileHandler("/tmp/mt5_mcp_server.log", mode="a"), ], ) log = logging.getLogger("mt5-mcp") # Sekunden, die auf eine Antwort aus dem Wine-Python gewartet wird RPYC_TIMEOUT = int(os.getenv("RPYC_TIMEOUT", "120")) # --------------------------------------------------------------------------- # MT5 Bridge (abstraction over mock / live) # --------------------------------------------------------------------------- class MT5Bridge: """Base class – subclassed by MockBridge and LiveBridge.""" def __init__(self): self.connected = False self.mode = "base" async def connect(self) -> bool: raise NotImplementedError async def disconnect(self): self.connected = False async def account_info(self) -> Dict[str, Any]: raise NotImplementedError async def symbols(self) -> List[str]: raise NotImplementedError async def symbol_info(self, symbol: str) -> Dict[str, Any]: raise NotImplementedError async def ohlc(self, symbol: str, timeframe: str, bars: int) -> List[Dict]: raise NotImplementedError async def tick(self, symbol: str) -> Dict[str, Any]: raise NotImplementedError async def positions(self) -> List[Dict]: raise NotImplementedError async def orders(self) -> List[Dict]: raise NotImplementedError # ---- Mock Bridge ---------------------------------------------------------- class MockBridge(MT5Bridge): """Generates realistic-looking fake data for testing.""" SYMBOLS = [ "EURUSD", "GBPUSD", "USDJPY", "USDCHF", "AUDUSD", "USDCAD", "NZDUSD", "XAUUSD", "XAGUSD", "BTCUSD", ] BASE_PRICES = { "EURUSD": 1.0850, "GBPUSD": 1.2650, "USDJPY": 149.50, "USDCHF": 0.8820, "AUDUSD": 0.6550, "USDCAD": 1.3580, "NZDUSD": 0.6120, "XAUUSD": 2650.00, "XAGUSD": 31.50, "BTCUSD": 67500.00, } SPREADS = { "EURUSD": 0.00012, "GBPUSD": 0.00015, "USDJPY": 0.015, "USDCHF": 0.00014, "AUDUSD": 0.00013, "USDCAD": 0.00016, "NZDUSD": 0.00015, "XAUUSD": 0.30, "XAGUSD": 0.03, "BTCUSD": 25.0, } def __init__(self): super().__init__() self.mode = "mock" async def connect(self) -> bool: self.connected = True log.info("MockBridge connected (simulated data)") return True async def account_info(self) -> Dict[str, Any]: return { "login": 12345678, "server": "MockBroker-Demo", "balance": 10000.00, "equity": 10025.50, "margin": 125.00, "free_margin": 9900.50, "margin_level": 8020.40, "leverage": 500, "currency": "USD", "name": "Demo Account", "company": "MockBroker Ltd", "trade_mode": "DEMO", } async def symbols(self) -> List[str]: return self.SYMBOLS async def symbol_info(self, symbol: str) -> Dict[str, Any]: if symbol not in self.BASE_PRICES: raise HTTPException(404, f"Symbol {symbol} not found") price = self.BASE_PRICES[symbol] spread = self.SPREADS.get(symbol, 0.0002) digits = 5 if price < 10 else (3 if price < 200 else (2 if price < 5000 else 1)) return { "symbol": symbol, "bid": round(price - spread / 2, digits), "ask": round(price + spread / 2, digits), "spread": spread, "digits": digits, "point": 10 ** -digits, "trade_contract_size": 100000 if "USD" in symbol and "XA" not in symbol else (100 if "XAU" in symbol else 1), "volume_min": 0.01, "volume_max": 100.0, "volume_step": 0.01, "description": f"{symbol} mock data", } async def ohlc(self, symbol: str, timeframe: str, bars: int, start_pos: int = 0) -> List[Dict]: """Generate mock OHLC bars.""" if symbol not in self.BASE_PRICES: raise HTTPException(404, f"Symbol {symbol} not found") tf_seconds = { "S5": 5, "S10": 10, "S15": 15, "S30": 30, "M1": 60, "M5": 300, "M15": 900, "M30": 1800, "H1": 3600, "H4": 14400, "D1": 86400, "W1": 604800, } seconds = tf_seconds.get(timeframe, 3600) price = self.BASE_PRICES[symbol] volatility = price * (0.0002 if seconds < 60 else 0.001) now = datetime.now(tz=None).replace(microsecond=0) # Align to timeframe now = now - timedelta(seconds=int(now.timestamp()) % max(seconds, 1)) data = [] current_price = price for i in range(bars - 1, -1, -1): bar_time = now - timedelta(seconds=seconds * i) o = current_price h = o + random.uniform(0, volatility) l = o - random.uniform(0, volatility) c = random.uniform(l, h) v = random.randint(50, 5000) digits = 5 if price < 10 else (3 if price < 200 else 2) data.append({ "time": bar_time.isoformat() + "Z", "timestamp": int(bar_time.timestamp()), "open": round(o, digits), "high": round(h, digits), "low": round(l, digits), "close": round(c, digits), "volume": v, }) current_price = c return data async def tick(self, symbol: str) -> Dict[str, Any]: if symbol not in self.BASE_PRICES: raise HTTPException(404, f"Symbol {symbol} not found") price = self.BASE_PRICES[symbol] spread = self.SPREADS.get(symbol, 0.0002) jitter = price * random.uniform(-0.0005, 0.0005) digits = 5 if price < 10 else (3 if price < 200 else 2) bid = round(price + jitter - spread / 2, digits) ask = round(price + jitter + spread / 2, digits) return { "symbol": symbol, "bid": bid, "ask": ask, "last": round((bid + ask) / 2, digits), "volume": random.randint(1, 100), "time": datetime.now(tz=None).isoformat() + "Z", } async def positions(self) -> List[Dict]: return [ { "ticket": 100001, "symbol": "EURUSD", "type": "BUY", "volume": 0.10, "open_price": 1.0840, "current_price": 1.0852, "profit": 12.00, "swap": -0.50, "sl": 1.0800, "tp": 1.0900, "open_time": (datetime.now(tz=None) - timedelta(hours=3)).isoformat() + "Z", "magic": 0, "comment": "mock position", } ] async def orders(self) -> List[Dict]: return [] # ---- Live Bridge (mt5linux) ----------------------------------------------- class LiveBridge(MT5Bridge): """Connects to real MT5 via mt5linux rpyc bridge.""" def __init__(self, host: str = "localhost", port: int = 8001): super().__init__() self.mode = "live" self.rpyc_host = host self.rpyc_port = port self.mt5 = None self._rpyc_conn = None # persistent rpyc connection for tick data def _get_rpyc(self): """Get or create persistent rpyc connection for direct Wine-Python access.""" import rpyc if self._rpyc_conn is None or self._rpyc_conn.closed: try: self._rpyc_conn = rpyc.classic.connect(self.rpyc_host, self.rpyc_port) self._rpyc_conn._config['sync_request_timeout'] = RPYC_TIMEOUT self._rpyc_conn.execute("import MetaTrader5 as mt5") self._rpyc_conn.execute("from datetime import datetime, timedelta") self._rpyc_conn.execute("mt5.initialize()") log.info("Direct rpyc connection established for tick data") except Exception as e: log.error(f"rpyc connection failed: {e}") self._rpyc_conn = None raise return self._rpyc_conn async def connect(self) -> bool: try: from mt5linux import MetaTrader5 as MT5 self.mt5 = MT5(host=self.rpyc_host, port=self.rpyc_port) # initialize() mit Login startet/verbindet das Terminal unter Wine und # dauert oft >30s (RPyC-Default) -> sonst "result expired" self.mt5._MetaTrader5__conn._config['sync_request_timeout'] = RPYC_TIMEOUT # Broker-Login aus Umgebungsvariablen (bots.env) – leer = Terminal-Login verwenden kwargs = {} if os.getenv("MT5_LOGIN"): kwargs = { "login": int(os.getenv("MT5_LOGIN")), "password": os.getenv("MT5_PASSWORD", ""), "server": os.getenv("MT5_SERVER", ""), } log.info(f"Logging in to MT5 account {kwargs['login']} on {kwargs['server']}") if not self.mt5.initialize(**kwargs): error = self.mt5.last_error() log.error(f"MT5 initialize failed: {error}") return False info = self.mt5.account_info() log.info(f"LiveBridge connected: login={info.login}, server={info.server}") self.connected = True return True except ImportError as e: # Kann auch von der Windows-Seite kommen (z.B. MetaTrader5 vs. numpy 2) log.error(f"ImportError: {e}") return False except Exception as e: log.error(f"LiveBridge connection failed: {e}") return False async def disconnect(self): if self._rpyc_conn: try: self._rpyc_conn.close() except Exception: pass self._rpyc_conn = None if self.mt5: try: self.mt5.shutdown() except Exception: pass self.connected = False async def account_info(self) -> Dict[str, Any]: if not self.mt5: raise HTTPException(503, "MT5 not connected") info = self.mt5.account_info() if info is None: raise HTTPException(500, f"account_info failed: {self.mt5.last_error()}") return { "login": info.login, "server": info.server, "balance": info.balance, "equity": info.equity, "margin": info.margin, "free_margin": info.margin_free, "margin_level": info.margin_level, "leverage": info.leverage, "currency": info.currency, "name": info.name, "company": info.company, "trade_mode": "DEMO" if info.trade_mode == 0 else "REAL", } async def symbols(self) -> List[str]: if not self.mt5: raise HTTPException(503, "MT5 not connected") syms = self.mt5.symbols_get() if syms is None: return [] return [s.name for s in syms if s.visible] async def symbol_info(self, symbol: str) -> Dict[str, Any]: if not self.mt5: raise HTTPException(503, "MT5 not connected") info = self.mt5.symbol_info(symbol) if info is None: raise HTTPException(404, f"Symbol {symbol} not found") return { "symbol": info.name, "bid": info.bid, "ask": info.ask, "spread": info.spread, "digits": info.digits, "point": info.point, "trade_contract_size": info.trade_contract_size, "volume_min": info.volume_min, "volume_max": info.volume_max, "volume_step": info.volume_step, "description": info.description, } async def ohlc(self, symbol: str, timeframe: str, bars: int, start_pos: int = 0) -> List[Dict]: if not self.mt5: raise HTTPException(503, "MT5 not connected") import pandas as pd # Check if seconds timeframe seconds_map = {"S5": 5, "S10": 10, "S15": 15, "S30": 30} if timeframe in seconds_map: return await self._ohlc_from_ticks(symbol, seconds_map[timeframe], bars) tf_map = { "M1": 1, "M5": 5, "M15": 15, "M30": 30, "H1": 16385, "H4": 16388, "D1": 16408, "W1": 32769, } tf = tf_map.get(timeframe, 16385) # default H1 rates = self.mt5.copy_rates_from_pos(symbol, tf, start_pos, bars) if rates is None or len(rates) == 0: raise HTTPException(404, f"No data for {symbol} {timeframe}") df = pd.DataFrame(rates) df["time"] = pd.to_datetime(df["time"], unit="s") result = [] for _, row in df.iterrows(): result.append({ "time": row["time"].isoformat() + "Z", "timestamp": int(row["time"].timestamp()), "open": float(row["open"]), "high": float(row["high"]), "low": float(row["low"]), "close": float(row["close"]), "volume": int(row["tick_volume"]), }) return result async def ohlc_range(self, symbol: str, timeframe: str, from_ts: int, to_ts: int) -> List[Dict]: """ Hole OHLCV Bars für einen Zeitraum via copy_rates_range. Ermöglicht chunked Downloads für beliebig lange History. Args: symbol: z.B. "XAUUSD" timeframe: M1, M5, M15, etc. from_ts: Unix timestamp Start to_ts: Unix timestamp Ende """ if not self.mt5: raise HTTPException(503, "MT5 not connected") import pandas as pd from datetime import datetime, timezone tf_map = { "M1": 1, "M5": 5, "M15": 15, "M30": 30, "H1": 16385, "H4": 16388, "D1": 16408, "W1": 32769, } tf = tf_map.get(timeframe, 1) # Convert timestamps to datetime for MT5 date_from = datetime.fromtimestamp(from_ts, tz=timezone.utc) date_to = datetime.fromtimestamp(to_ts, tz=timezone.utc) try: rates = self.mt5.copy_rates_range(symbol, tf, date_from, date_to) except Exception as e: # rpyc datetime workaround: pass as naive datetime try: date_from_naive = datetime.utcfromtimestamp(from_ts) date_to_naive = datetime.utcfromtimestamp(to_ts) rates = self.mt5.copy_rates_range(symbol, tf, date_from_naive, date_to_naive) except Exception as e2: raise HTTPException(500, f"copy_rates_range failed: {e2}") if rates is None or len(rates) == 0: return [] df = pd.DataFrame(rates) df["time"] = pd.to_datetime(df["time"], unit="s") result = [] for _, row in df.iterrows(): result.append({ "time": row["time"].isoformat() + "Z", "timestamp": int(row["time"].timestamp()), "open": float(row["open"]), "high": float(row["high"]), "low": float(row["low"]), "close": float(row["close"]), "volume": int(row["tick_volume"]), }) return result async def _ohlc_from_ticks(self, symbol: str, period_seconds: int, bars: int) -> List[Dict]: """Build sub-minute OHLC bars from real tick data via persistent rpyc.""" import pandas as pd conn = self._get_rpyc() # Fetch ticks — all in Wine-Python, only pull converted tuples ticks_needed = max(bars * 20, 5000) minutes_back = max(int(bars * period_seconds / 60) * 2, 10) conn.execute( f'_ticks = mt5.copy_ticks_from("{symbol}", ' f'datetime.utcnow() - timedelta(minutes={minutes_back}), ' f'{ticks_needed}, mt5.COPY_TICKS_ALL)' ) count = conn.eval("len(_ticks)") log.info(f"Fetched {count} ticks for {symbol} ({minutes_back}min back)") if count == 0: raise HTTPException(404, f"No tick data for {symbol} — market may be closed") # Convert on Wine side, pull as plain list conn.execute( "_td = [(int(_ticks[i][0]), float(_ticks[i][1]), float(_ticks[i][2])) " "for i in range(len(_ticks))]" ) tick_data = list(conn.eval("_td")) # Build DataFrame locally df = pd.DataFrame(tick_data, columns=["time", "bid", "ask"]) df["dt"] = pd.to_datetime(df["time"], unit="s") df["mid"] = (df["bid"] + df["ask"]) / 2 # Resample into bars rule = f"{period_seconds}s" grouped = df.set_index("dt")["mid"].resample(rule) ohlc_open = grouped.first().dropna() ohlc_high = grouped.max().dropna() ohlc_low = grouped.min().dropna() ohlc_close = grouped.last().dropna() vol = grouped.count() idx = ohlc_open.index.intersection(ohlc_close.index) idx = idx.intersection(ohlc_high.index).intersection(ohlc_low.index) idx = idx[-bars:] if len(idx) == 0: raise HTTPException(404, f"No complete {period_seconds}s bars for {symbol}") # Get current tick to fix last (incomplete) bar try: conn.execute(f'_curtick = mt5.symbol_info_tick("{symbol}")') cur_bid = conn.eval("float(_curtick.bid)") cur_ask = conn.eval("float(_curtick.ask)") cur_mid = (cur_bid + cur_ask) / 2 except Exception: cur_mid = None result = [] last_idx = idx[-1] for ts in idx: o = float(ohlc_open[ts]) h = float(ohlc_high[ts]) l = float(ohlc_low[ts]) c = float(ohlc_close[ts]) # Fix last bar with current price if ts == last_idx and cur_mid is not None: c = cur_mid if cur_mid > h: h = cur_mid if cur_mid < l: l = cur_mid result.append({ "time": ts.isoformat() + "Z", "timestamp": int(ts.timestamp()), "open": round(o, 6), "high": round(h, 6), "low": round(l, 6), "close": round(c, 6), "volume": int(vol.get(ts, 0)), }) log.info(f"Built {len(result)} x {period_seconds}s bars from {len(df)} ticks") return result async def order_open(self, symbol: str, side: str, volume: float, sl: float = 0, tp: float = 0, comment: str = "") -> Dict[str, Any]: """Open a market order.""" if not self.mt5: raise HTTPException(503, "MT5 not connected") # Get symbol info for filling info = self.mt5.symbol_info(symbol) if info is None: raise HTTPException(404, f"Symbol {symbol} not found") price = info.ask if side == "buy" else info.bid order_type = 0 if side == "buy" else 1 # ORDER_TYPE_BUY / SELL request = { "action": 1, # TRADE_ACTION_DEAL "symbol": symbol, "volume": volume, "type": order_type, "price": price, "sl": sl, "tp": tp, "deviation": 20, "magic": 202604, "comment": comment or "xgboost_sweep", "type_time": 0, # ORDER_TIME_GTC "type_filling": 1, # ORDER_FILLING_IOC } result = self.mt5.order_send(request) if result is None: error = self.mt5.last_error() raise HTTPException(500, f"order_send failed: {error}") return { "retcode": result.retcode, "deal": result.deal, "order": result.order, "volume": result.volume, "price": result.price, "comment": result.comment, "success": result.retcode == 10009, # TRADE_RETCODE_DONE "error": result.comment if result.retcode != 10009 else None, } async def order_modify(self, ticket: int, sl: float = None, tp: float = None) -> Dict[str, Any]: """Modify SL/TP of an existing position.""" if not self.mt5: raise HTTPException(503, "MT5 not connected") # Get current position positions = self.mt5.positions_get(ticket=ticket) if positions is None or len(positions) == 0: raise HTTPException(404, f"Position {ticket} not found") pos = positions[0] request = { "action": 6, # TRADE_ACTION_SLTP "position": ticket, "symbol": pos.symbol, "sl": sl if sl is not None else pos.sl, "tp": tp if tp is not None else pos.tp, "type_filling": 1, # ORDER_FILLING_IOC (Vantage) } result = self.mt5.order_send(request) if result is None: error = self.mt5.last_error() raise HTTPException(500, f"order_modify failed: {error}") return { "retcode": result.retcode, "success": result.retcode == 10009, "error": result.comment if result.retcode != 10009 else None, } async def order_close(self, ticket: int) -> Dict[str, Any]: """Close an existing position.""" if not self.mt5: raise HTTPException(503, "MT5 not connected") positions = self.mt5.positions_get(ticket=ticket) if positions is None or len(positions) == 0: raise HTTPException(404, f"Position {ticket} not found") pos = positions[0] close_type = 1 if pos.type == 0 else 0 # Reverse: BUY->SELL, SELL->BUY price = self.mt5.symbol_info_tick(pos.symbol) close_price = price.bid if pos.type == 0 else price.ask request = { "action": 1, # TRADE_ACTION_DEAL "symbol": pos.symbol, "volume": pos.volume, "type": close_type, "position": ticket, "price": close_price, "deviation": 20, "magic": 202604, "comment": "xgboost_close", "type_time": 0, "type_filling": 1, } result = self.mt5.order_send(request) if result is None: error = self.mt5.last_error() raise HTTPException(500, f"order_close failed: {error}") return { "retcode": result.retcode, "deal": result.deal, "success": result.retcode == 10009, "error": result.comment if result.retcode != 10009 else None, } async def tick(self, symbol: str) -> Dict[str, Any]: if not self.mt5: raise HTTPException(503, "MT5 not connected") tick = self.mt5.symbol_info_tick(symbol) if tick is None: raise HTTPException(404, f"No tick for {symbol}") return { "symbol": symbol, "bid": tick.bid, "ask": tick.ask, "last": tick.last, "volume": tick.volume, "time": datetime.fromtimestamp(tick.time, tz=timezone.utc).isoformat() + "Z", } async def positions(self) -> List[Dict]: if not self.mt5: raise HTTPException(503, "MT5 not connected") pos = self.mt5.positions_get() if pos is None: return [] result = [] for p in pos: result.append({ "ticket": p.ticket, "symbol": p.symbol, "type": "BUY" if p.type == 0 else "SELL", "volume": p.volume, "open_price": p.price_open, "current_price": p.price_current, "profit": p.profit, "swap": p.swap, "sl": p.sl, "tp": p.tp, "open_time": datetime.fromtimestamp(p.time, tz=timezone.utc).isoformat() + "Z", "magic": p.magic, "comment": p.comment, }) return result async def orders(self) -> List[Dict]: if not self.mt5: raise HTTPException(503, "MT5 not connected") ords = self.mt5.orders_get() if ords is None: return [] return [ { "ticket": o.ticket, "symbol": o.symbol, "type": o.type, "volume": o.volume_current, "price": o.price_open, "sl": o.sl, "tp": o.tp, "time_setup": datetime.fromtimestamp(o.time_setup, tz=timezone.utc).isoformat() + "Z", "comment": o.comment, } for o in ords ] # --------------------------------------------------------------------------- # Global bridge instance (set in lifespan) # --------------------------------------------------------------------------- bridge: MT5Bridge = MockBridge() @asynccontextmanager async def lifespan(app: FastAPI): global bridge ok = await bridge.connect() # Im Container startet MT5 beim ersten Mal mehrere Minuten – so lange warten statt # dauerhaft "degraded" zu laufen. Healthcheck bleibt rot bis die Verbindung steht. while not ok and bridge.mode == "live": log.warning("MT5 not ready yet – retrying in 15s") await asyncio.sleep(15) ok = await bridge.connect() if ok: log.info(f"Server starting in {bridge.mode.upper()} mode") else: log.warning(f"Bridge connect failed ({bridge.mode}) – server runs but data may be unavailable") yield await bridge.disconnect() log.info("Server shutdown complete") # --------------------------------------------------------------------------- # FastAPI App # --------------------------------------------------------------------------- app = FastAPI( title="MT5 MCP Server", version="3.0.0", description="Linux MCP server for MetaTrader 5 – mock or live mode", lifespan=lifespan, ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) start_time = datetime.now(tz=None) # --------------------------------------------------------------------------- # Endpoints # --------------------------------------------------------------------------- @app.get("/") async def root(): return { "service": "MT5 MCP Server", "version": "3.0.0", "mode": bridge.mode, "connected": bridge.connected, "uptime_s": int((datetime.now(tz=None) - start_time).total_seconds()), "endpoints": [ "GET /health", "GET /account", "GET /symbols", "GET /symbol/{symbol}", "GET /ohlc/{symbol}?timeframe=H1&bars=100", "GET /tick/{symbol}", "GET /positions", "GET /orders", "GET /mcp/tools", ], } @app.get("/health") async def health(): return { "status": "ok" if bridge.connected else "degraded", "mode": bridge.mode, "connected": bridge.connected, "uptime_s": int((datetime.now(tz=None) - start_time).total_seconds()), "timestamp": datetime.now(tz=None).isoformat() + "Z", } @app.get("/account") async def account(): return await bridge.account_info() @app.get("/symbols") async def symbols(): syms = await bridge.symbols() return {"count": len(syms), "symbols": syms} @app.get("/symbol/{symbol}") async def symbol_info(symbol: str): return await bridge.symbol_info(symbol.upper()) @app.get("/ohlc/{symbol}") async def ohlc( symbol: str, timeframe: str = Query("H1", pattern="^(S5|S10|S15|S30|M1|M5|M15|M30|H1|H4|D1|W1)$"), bars: int = Query(100, ge=1, le=10000), start: int = Query(0, ge=0, description="Offset: 0=latest, 10000=next page back"), ): data = await bridge.ohlc(symbol.upper(), timeframe, bars, start_pos=start) return { "symbol": symbol.upper(), "timeframe": timeframe, "bars": len(data), "data": data, } @app.get("/ohlc_range/{symbol}") async def ohlc_range( symbol: str, timeframe: str = Query("M1", pattern="^(M1|M5|M15|M30|H1|H4|D1|W1)$"), from_ts: int = Query(..., description="Unix timestamp start"), to_ts: int = Query(..., description="Unix timestamp end"), ): """Fetch OHLCV bars for a specific time range (chunked downloads).""" data = await bridge.ohlc_range(symbol.upper(), timeframe, from_ts, to_ts) return { "symbol": symbol.upper(), "timeframe": timeframe, "from": from_ts, "to": to_ts, "bars": len(data), "data": data, } # --------------------------------------------------------------------------- # Order Execution Endpoints # --------------------------------------------------------------------------- class OrderRequest(BaseModel): symbol: str side: str # "buy" or "sell" volume: float = 0.01 sl: float = 0 tp: float = 0 comment: str = "" class ModifyRequest(BaseModel): ticket: int sl: float = None tp: float = None class CloseRequest(BaseModel): ticket: int @app.post("/order/open") async def order_open(req: OrderRequest): """Open a market order.""" if req.side not in ("buy", "sell"): raise HTTPException(400, "side must be 'buy' or 'sell'") return await bridge.order_open( req.symbol.upper(), req.side, req.volume, req.sl, req.tp, req.comment ) @app.post("/order/modify") async def order_modify(req: ModifyRequest): """Modify SL/TP of existing position.""" return await bridge.order_modify(req.ticket, req.sl, req.tp) @app.post("/order/close") async def order_close(req: CloseRequest): """Close existing position.""" return await bridge.order_close(req.ticket) @app.get("/tick/{symbol}") async def tick(symbol: str): return await bridge.tick(symbol.upper()) @app.get("/positions") async def positions(): pos = await bridge.positions() return {"count": len(pos), "positions": pos} @app.get("/orders") async def orders(): ords = await bridge.orders() return {"count": len(ords), "orders": ords} # --------------------------------------------------------------------------- # MCP Tool Definitions (for AI integration / n8n) # --------------------------------------------------------------------------- @app.get("/mcp/tools") async def mcp_tools(): """Return MCP-compatible tool definitions.""" return { "tools": [ { "name": "get_account_info", "description": "Get MT5 trading account info (balance, equity, margin, leverage)", "inputSchema": {"type": "object", "properties": {}}, }, { "name": "get_symbols", "description": "List available trading symbols", "inputSchema": {"type": "object", "properties": {}}, }, { "name": "get_symbol_info", "description": "Get details for a specific symbol (bid, ask, spread, contract size)", "inputSchema": { "type": "object", "properties": {"symbol": {"type": "string", "description": "Symbol name, e.g. EURUSD"}}, "required": ["symbol"], }, }, { "name": "get_ohlc", "description": "Get OHLC candlestick data for a symbol", "inputSchema": { "type": "object", "properties": { "symbol": {"type": "string", "description": "Symbol name"}, "timeframe": {"type": "string", "enum": ["M1", "M5", "M15", "M30", "H1", "H4", "D1", "W1"]}, "bars": {"type": "integer", "description": "Number of bars (1-10000)", "default": 100}, }, "required": ["symbol"], }, }, { "name": "get_tick", "description": "Get current tick (bid/ask) for a symbol", "inputSchema": { "type": "object", "properties": {"symbol": {"type": "string"}}, "required": ["symbol"], }, }, { "name": "get_positions", "description": "Get open trading positions", "inputSchema": {"type": "object", "properties": {}}, }, { "name": "get_orders", "description": "Get pending orders", "inputSchema": {"type": "object", "properties": {}}, }, ] } # --------------------------------------------------------------------------- # Main # --------------------------------------------------------------------------- def main(): global bridge parser = argparse.ArgumentParser(description="MT5 MCP Server") parser.add_argument("--mock", action="store_true", help="Use mock data (default)") parser.add_argument("--live", action="store_true", help="Connect to real MT5 via mt5linux") parser.add_argument("--host", default="0.0.0.0", help="Listen host (default: 0.0.0.0)") parser.add_argument("--port", type=int, default=8000, help="Listen port (default: 8000)") parser.add_argument("--rpyc-host", default="localhost", help="mt5linux rpyc host") parser.add_argument("--rpyc-port", type=int, default=8001, help="mt5linux rpyc port") args = parser.parse_args() if args.live: bridge = LiveBridge(host=args.rpyc_host, port=args.rpyc_port) else: bridge = MockBridge() log.info(f"Starting MT5 MCP Server on {args.host}:{args.port} [{bridge.mode} mode]") uvicorn.run(app, host=args.host, port=args.port, log_level="info") if __name__ == "__main__": main()