metals-mcp/src/metals_mcp/mcp_tools.py

from __future__ import annotations

from typing import Any

from .config import DB_PATH
from .storage import last_candle, latest_candles, stats
from .swissquote import SwissquoteClient

client = SwissquoteClient()


def get_capabilities() -> dict[str, Any]:
    return {
        "server": "metals-mcp",
        "transport": "fastmcp+sse",
        "tools": [
            "get_price",
            "get_ohlcv",
            "get_indicator",
            "get_market_snapshot",
            "get_top_movers",
            "get_capabilities",
            "get_regime",
        ],
        "backend": "swissquote-poller+sqlite",
        "database": str(DB_PATH),
    }


def get_price(symbol: str) -> dict[str, Any]:
    quote = client.fetch_quote(symbol)
    if not quote:
        return {"symbol": symbol, "price": None, "timestamp": None, "source": "swissquote", "status": "unavailable"}
    return {"symbol": symbol, "price": quote.mid, "timestamp": quote.timestamp, "source": "swissquote"}


def get_ohlcv(symbol: str, timeframe: str = "5m", limit: int = 100) -> dict[str, Any]:
    candles = latest_candles(DB_PATH, symbol.upper(), timeframe, limit)
    return {"symbol": symbol.upper(), "timeframe": timeframe, "limit": limit, "candles": candles}


def get_candles(symbol: str, timeframe: str = "5m", limit: int = 100) -> dict[str, Any]:
    return get_ohlcv(symbol, timeframe=timeframe, limit=limit)


def get_last_candle(symbol: str, timeframe: str = "5m") -> dict[str, Any]:
    return {"symbol": symbol.upper(), "timeframe": timeframe, "candle": last_candle(DB_PATH, symbol.upper(), timeframe)}


def get_indicator(symbol: str, indicator: str, timeframe: str = "5m", params: dict[str, Any] | None = None) -> dict[str, Any]:
    return {"symbol": symbol, "indicator": indicator, "timeframe": timeframe, "params": params or {}, "value": None, "status": "scaffolded"}


def get_market_snapshot(symbol: str) -> dict[str, Any]:
    candle = last_candle(DB_PATH, symbol.upper(), "5m")
    price = candle["close"] if candle else None
    return {"symbol": symbol.upper(), "price": price, "trend_bias": "range", "status": "scaffolded"}


def get_top_movers(limit: int = 10) -> dict[str, Any]:
    return {"limit": limit, "movers": [], "status": "scaffolded"}


def get_regime(symbol: str, timeframe: str = "5m") -> dict[str, Any]:
    candles = latest_candles(DB_PATH, symbol.upper(), timeframe, 20)
    return {"symbol": symbol.upper(), "timeframe": timeframe, "candles": candles, "regime": None, "status": "scaffolded"}


def get_health() -> dict[str, Any]:
    try:
        store = stats(DB_PATH)
    except Exception:
        store = {"ticks": 0, "candles": 0}
    return {"ok": True, "store": store}