crypto-mcp/server_fastmcp.py

"""FastMCP transport wrapper for crypto tools (compliance-first)."""

from fastapi import FastAPI
from mcp.server.fastmcp import FastMCP
from mcp.server.transport_security import TransportSecuritySettings

import services
from mcp_tools import MCP_TOOLS
from cache import get_cache_stats


mcp = FastMCP(
    "crypto-mcp",
    transport_security=TransportSecuritySettings(
        enable_dns_rebinding_protection=False,
    ),
)


@mcp.tool(description="Get the current USD price of a cryptocurrency.")
async def get_price(symbol: str):
    return await services.get_price(symbol)


@mcp.tool(description="Get OHLCV candlestick data for a crypto asset.")
async def get_ohlcv(symbol: str, timeframe: str = "1h", limit: int = 100):
    return await services.get_ohlcv(symbol, timeframe, limit)


@mcp.tool(description="Compute a technical indicator for a crypto asset.")
async def get_indicator(symbol: str, indicator: str, timeframe: str = "1h", params: dict | None = None):
    return await services.get_indicator(symbol, indicator, timeframe, params or {})


@mcp.tool(description="Get a compact market snapshot for a crypto asset.")
async def get_market_snapshot(symbol: str):
    return await services.get_market_snapshot(symbol)


@mcp.tool(description="Get top gaining and losing crypto assets by 24h % change.")
async def get_top_movers(limit: int = 10):
    return await services.get_top_movers(limit)


app = FastAPI(title="Crypto MCP Server")
app.mount("/mcp", mcp.sse_app())


@app.get("/")
def root():
    return {"status": "ok", "transport": "fastmcp+sse", "mount": "/mcp", "tools": [t["name"] for t in MCP_TOOLS]}


@app.get("/health")
def health():
    return {"status": "ok", "cache": get_cache_stats(), "tools": [t["name"] for t in MCP_TOOLS]}