from __future__ import annotations from contextlib import asynccontextmanager import asyncio from fastapi import FastAPI, Form, HTTPException from fastapi.responses import HTMLResponse, RedirectResponse from fastmcp import FastMCP from .models import AccountView from . import repo from .services_bitstamp import fetch_account_info as fetch_remote_account_info, fetch_account_fees as fetch_remote_account_fees from .bitstamp_metadata import METADATA_REFRESH_SECONDS, refresh_metadata, list_markets as bitstamp_list_markets, list_currencies as bitstamp_list_currencies from .bitstamp_fx import FX_REFRESH_SECONDS, refresh_eur_usd from .bitstamp_private_ws import private_ws_main from .services_orders import place_order as service_place_order, query_order as service_query_order, cancel_order as service_cancel_order, get_open_orders as service_get_open_orders, cancel_all_orders as service_cancel_all_orders from .storage import init_db mcp = FastMCP("exec-mcp") async def _metadata_refresh_loop() -> None: while True: try: refresh_metadata() except Exception: pass await asyncio.sleep(METADATA_REFRESH_SECONDS) async def _fx_refresh_loop() -> None: while True: try: refresh_eur_usd() except Exception: pass await asyncio.sleep(FX_REFRESH_SECONDS) @asynccontextmanager async def lifespan(_: FastAPI): init_db() try: refresh_metadata() except Exception: pass stop_event = asyncio.Event() metadata_task = asyncio.create_task(_metadata_refresh_loop()) fx_task = asyncio.create_task(_fx_refresh_loop()) private_ws_task = asyncio.create_task(private_ws_main(stop_event)) try: yield finally: stop_event.set() metadata_task.cancel() fx_task.cancel() private_ws_task.cancel() app = FastAPI(title="exec-mcp", lifespan=lifespan) app.mount("/mcp", mcp.http_app(path="/sse", transport="sse")) SUPPORTED_VENUES = {"bitstamp"} REPORT_ORDER_STATUSES = {"all", "open", "cancelled", "finished"} def _normalize_report_status(status: str) -> str: value = str(status or "").strip().lower() if value not in REPORT_ORDER_STATUSES: raise HTTPException(status_code=400, detail="status must be one of all, open, cancelled, finished") return value def _normalize_report_limit(limit: int) -> int: try: value = int(limit) except Exception as exc: raise HTTPException(status_code=400, detail="limit must be an integer") from exc if value < 1: raise HTTPException(status_code=400, detail="limit must be at least 1") if value > 500: raise HTTPException(status_code=400, detail="limit must be at most 500") return value @app.get("/", response_class=HTMLResponse) def http_root() -> str: return """

exec-mcp

Execution layer for Trader27, with a dashboard for account management.

""" @app.get("/health") def http_health() -> dict: return {"ok": True, "server": "exec-mcp"} @app.get("/dashboard", response_class=HTMLResponse) def http_dashboard() -> str: rows = list_accounts(enabled_only=False) options = "".join(f'' for v in sorted(SUPPORTED_VENUES)) table_rows = "".join( f""" {row['display_name'] or ''} {row['venue'] or ''} {row['venue_account_ref'] or ''} {row['description'] or ''} {'yes' if row['enabled'] else 'no'}
""" for row in rows ) return f"""

exec-mcp dashboard

Create account

Accounts

{table_rows}
namevenueexchange account refdescriptionenabledactions
""" @app.post("/dashboard/accounts/create") def http_dashboard_create_account( display_name: str = Form(""), venue: str = Form(...), venue_account_ref: str = Form(""), api_key: str = Form(...), api_secret: str = Form(...), description: str | None = Form(None), enabled: bool = Form(False), ) -> RedirectResponse: repo.create_account( display_name=display_name, venue=venue, venue_account_ref=venue_account_ref, api_key=api_key, api_secret=api_secret, description=description, enabled=enabled, ) return RedirectResponse(url="/dashboard", status_code=303) @app.get("/dashboard/accounts/{account_id}/edit", response_class=HTMLResponse) def http_dashboard_edit_account(account_id: str) -> str: row = repo.get_account(account_id) checked = "checked" if row["enabled"] else "" return f"""

Edit account

Back to dashboard

""" @app.post("/dashboard/accounts/{account_id}/update") def http_dashboard_update_account( account_id: str, display_name: str = Form(""), description: str = Form(""), enabled: bool = Form(False), ) -> RedirectResponse: repo.update_account(account_id=account_id, display_name=display_name, description=description or None, enabled=enabled) return RedirectResponse(url="/dashboard", status_code=303) @app.post("/dashboard/accounts/{account_id}/delete") def http_dashboard_delete_account(account_id: str) -> RedirectResponse: repo.delete_account(account_id=account_id) return RedirectResponse(url="/dashboard", status_code=303) @mcp.tool() def list_accounts(enabled_only: bool = True, venue: str | None = None) -> list[dict]: """list_accounts(enabled_only=True, venue=None) Return account metadata only, never secrets. If `venue` is set, filter to that exchange. `enabled_only` defaults to true. """ return repo.list_accounts(venue=venue, enabled_only=enabled_only) @mcp.tool() def list_markets() -> list[dict]: """list_markets() Return cached Bitstamp market metadata from the local SQLite cache. """ return bitstamp_list_markets() @mcp.tool() def list_currencies() -> list[dict]: """list_currencies() Return cached Bitstamp currency metadata from the local SQLite cache. """ return bitstamp_list_currencies() @mcp.tool() def get_account_info(account_id: str) -> dict: """get_account_info(account_id) Return account metadata plus venue-specific live account details when available. """ account = repo.get_account(account_id) if account["venue"] == "bitstamp": return fetch_remote_account_info(account_id) raise HTTPException(status_code=400, detail="unsupported venue") @mcp.tool() def get_account_fees(account_id: str, market_symbol: str | None = None) -> dict: """get_account_fees(account_id, market_symbol=None) Return Bitstamp fee information for one account, optionally scoped to a market symbol. """ account = repo.get_account(account_id) if account["venue"] == "bitstamp": return fetch_remote_account_fees(account_id, market_symbol) raise HTTPException(status_code=400, detail="unsupported venue") @mcp.tool() def place_order(account_id: str, market: str, side: str, order_type: str, amount, price=None, expire_time: int | None = None, client_id: str | None = None, client_order_id: str | None = None) -> dict: """place_order(account_id, market, side, order_type, amount, price=None, expire_time=None, client_id=None, client_order_id=None) Place a Bitstamp order. `market` is a Bitstamp symbol like `xrpusd`, `amount` is base units, and `expire_time` is relative seconds from now. `client_id` is optional and must be a string when provided. """ return service_place_order(account_id=account_id, market=market, side=side, order_type=order_type, amount=amount, price=price, expire_time=expire_time, client_id=client_id, client_order_id=client_order_id) @mcp.tool() def query_order(account_id: str, order_id, client_order_id: str | None = None, omit_transactions: bool | None = None) -> dict: """query_order(account_id, order_id, client_order_id=None, omit_transactions=None) Query a Bitstamp order by exchange order id. `client_order_id` and `omit_transactions` are optional Bitstamp query parameters. """ return service_query_order(account_id=account_id, order_id=order_id, client_order_id=client_order_id, omit_transactions=omit_transactions) @mcp.tool() def cancel_order(account_id: str, order_id) -> dict: """cancel_order(account_id, order_id) Cancel one Bitstamp order by exchange order id. """ return service_cancel_order(account_id=account_id, order_id=order_id) @mcp.tool() def get_open_orders(account_id: str, client_id: str | None = None) -> dict: """get_open_orders(account_id, client_id=None) List open orders for one account. If `client_id` is provided, only return orders recorded with that client identifier. """ return service_get_open_orders(account_id=account_id, client_id=client_id) @mcp.tool() def cancel_all_orders(account_id: str, client_id: str | None = None) -> dict: """cancel_all_orders(account_id, client_id=None) Cancel all open orders for one account. If `client_id` is provided, only cancel orders recorded with that client identifier. """ return service_cancel_all_orders(account_id=account_id, client_id=client_id) @mcp.tool() def report_recent_trades(account_id: str, status: str = "finished", client_id: str | None = None, limit: int = 50) -> dict: """report_recent_trades(account_id, status="finished", client_id=None, limit=50) Return recent order snapshots for reporting, ordered newest first. Status may be `all`, `open`, `cancelled`, or `finished`. `client_id` is optional and filters to one reporting stream when provided. """ repo.get_account(account_id) status = _normalize_report_status(status) limit = _normalize_report_limit(limit) orders = repo.list_recent_orders(account_id=account_id, client_id=client_id, status=status, limit=limit) return { "ok": True, "account_id": account_id, "client_id": client_id, "status": status, "limit": limit, "count": len(orders), "orders": orders, } @mcp.tool() def report_balance_snapshots(account_id: str, limit: int = 50) -> list[dict]: """report_balance_snapshots(account_id, limit=50) Return stored wallet balance snapshots for the account, ordered newest first. """ repo.get_account(account_id) limit = _normalize_report_limit(limit) return repo.list_account_balance_snapshots(account_id=account_id, limit=limit) def main() -> None: init_db()