Scaffold metals MCP server

.env.example

@@ -0,0 +1,6 @@
+METALS_HOST=127.0.0.1
+METALS_PORT=8515
+METALS_DATA_DIR=./data
+METALS_LOG_DIR=./logs
+METALS_DB_PATH=./data/metals.sqlite3
+SWISSQUOTE_POLL_INTERVAL_SECONDS=0.2

.gitignore

@@ -0,0 +1,11 @@
+__pycache__/
+.venv/
+.pytest_cache/
+*.pyc
+logs/
+data/
+*.sqlite3
+*.db
+uvicorn.log
+server.log
+server.pid

Dockerfile

@@ -0,0 +1,7 @@
+FROM python:3.13-slim
+WORKDIR /app
+COPY requirements.txt ./
+RUN pip install --no-cache-dir -r requirements.txt
+COPY . .
+EXPOSE 8515
+CMD ["python", "main.py"]

PROJECT.md

@@ -0,0 +1,60 @@
+# PROJECT.md
+
+## Purpose
+
+`metals-mcp` is a compact MCP server for metals-related tools.
+The implementation lives under `src/metals_mcp/`.
+The public interface is intended to resemble `crypto-mcp` closely, while the backend uses Swissquote polling and a local candle store.
+The intended analytical shape is simple: clock-aligned **5m candles** as the base market view, with gold and silver as the primary focus.
+
+## Current interface
+
+- `GET /` → health + tool list
+- `GET /health` → health + store stats
+- FastMCP SSE transport mounted at `GET /mcp/sse`
+- Local default port: `8515`
+
+Tool calls are performed via FastMCP’s message transport under `/mcp/messages/`.
+
+## Tool set
+
+- `get_price`
+- `get_ohlcv`
+- `get_indicator`
+- `get_market_snapshot`
+- `get_top_movers`
+- `get_capabilities`
+- `get_regime`
+
+## Candle model
+
+- Base timeframe: `5m`
+- Candles are clock-aligned, e.g. `00:00–00:05`, `00:05–00:10`
+- Missing ticks inside a window do not break the candle, the window itself is the candle boundary
+- The system is for market orientation and regime reading, not execution
+
+## Done
+
+- Live Swissquote `get_price` for metals like `XAU`
+- Background poller that keeps the server alive and updating
+- Clock-aligned 5m candle storage in SQLite
+- Crypto-like tool naming where it helps client compatibility
+
+## Planned
+
+- Real regime calculations for metals
+- Retention policy for historical candles
+- Additional pair coverage only if it proves useful
+
+## Notes
+
+- Keep the transport small and predictable.
+- Mirror the crypto server’s MCP/HTTP shape where it helps client compatibility.
+- Keep Swissquote polling, storage, and candle aggregation behind the tools.
+
+## Verification
+
+- `./tests.sh`
+- `./run.sh`
+- `./killserver.sh`
+- `./restart.sh`

README.md

@@ -0,0 +1,63 @@
+# Metals MCP Server
+
+FastMCP-based metals market data server backed by a Swissquote poller and local candle store.
+
+## Transport
+
+- MCP SSE transport mounted at `/mcp`
+- SSE stream endpoint: `/mcp/sse`
+- message endpoint handled by FastMCP under `/mcp/messages/`
+- no legacy `/rpc` compatibility path
+
+## Runtime
+
+```bash
+source .venv/bin/activate
+pip install -r requirements.txt
+./run.sh
+```
+
+Default URL base: `http://127.0.0.1:8515`
+
+## HTTP
+
+- `GET /` → health + tool list
+- `GET /health` → health + cache/store stats
+
+## Tools
+
+- `get_price`
+- `get_candles`
+- `get_last_candle`
+- `get_capabilities`
+
+## Notes
+
+### Done
+- MCP surface mirrors `crypto-mcp` tool names where it makes sense.
+- `get_price` fetches live Swissquote quotes for metals like `XAU`.
+- An internal background poller keeps the server self-sufficient.
+- 5m candles are clock-aligned and persisted in SQLite.
+
+### Planned
+- Fill out `get_indicator`, `get_market_snapshot`, `get_top_movers`, and `get_regime` with real regime math.
+- Add retention cleanup for older candles.
+- Expand the watched pair set only if it proves useful for market orientation.
+
+### Design
+- The public surface should mirror `crypto-mcp` where practical.
+- Local default port is `8515`.
+- Under the hood, the server uses Swissquote polling plus a local candle store.
+- The repo now contains a working scaffold and the first live price path.
+
+## Verification
+
+```bash
+./tests.sh
+./run.sh
+```
+
+## Housekeeping
+
+- `./killserver.sh` stops stale listeners on the configured port
+- `./restart.sh` chains kill and run

discover_swissquote_api.py

@@ -0,0 +1,140 @@
+import requests
+import time
+import json
+import random
+from datetime import datetime, UTC
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from threading import Lock
+
+BASE_URL = "https://forex-data-feed.swissquote.com/public-quotes/bboquotes/instrument/{}"
+
+HEADERS = {
+    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0 Safari/537.36",
+    "Accept": "application/json, text/plain, */*",
+    "Accept-Language": "en-US,en;q=0.9",
+    "Connection": "keep-alive"
+}
+
+BASE_ASSETS = [
+    "XAU", "XAG", "XPT", "XPD",
+    "EUR", "USD", "GBP", "JPY", "CHF", "AUD", "CAD", "NZD"
+]
+
+QUOTE_ASSETS = [
+    "USD", "EUR", "JPY", "CHF", "GBP"
+]
+
+OUTPUT_FILE = "swissquote_pairs.json"
+
+# Rate limiting
+REQUESTS_PER_SECOND = 5
+MIN_INTERVAL = 1.0 / REQUESTS_PER_SECOND
+last_request_time = 0
+rate_lock = Lock()
+
+
+def rate_limited_request():
+    global last_request_time
+    with rate_lock:
+        now = time.time()
+        elapsed = now - last_request_time
+        if elapsed < MIN_INTERVAL:
+            time.sleep(MIN_INTERVAL - elapsed)
+        last_request_time = time.time()
+
+
+def fetch(symbol, retries=2):
+    url = BASE_URL.format(symbol)
+
+    for attempt in range(retries + 1):
+        try:
+            rate_limited_request()
+            time.sleep(random.uniform(0.02, 0.08))  # jitter
+
+            r = requests.get(url, headers=HEADERS, timeout=3)
+
+            if r.status_code != 200:
+                raise Exception(f"HTTP {r.status_code}")
+
+            data = r.json()
+
+            # Normalize response (dict or list)
+            if isinstance(data, list):
+                if not data:
+                    return None
+                data = data[0]
+
+            if not isinstance(data, dict):
+                return None
+
+            prices = data.get("spreadProfilePrices")
+            if not prices:
+                return None
+
+            p = prices[0]
+
+            bid = float(p.get("bid", 0))
+            ask = float(p.get("ask", 0))
+            ts = p.get("timestamp")
+
+            if bid > 0 and ask > 0:
+                return {
+                    "symbol": symbol,
+                    "bid": bid,
+                    "ask": ask,
+                    "timestamp": ts
+                }
+
+            return None
+
+        except Exception as e:
+            if attempt < retries:
+                time.sleep(0.3 * (2 ** attempt))  # backoff
+            else:
+                print(f"[FAIL] {symbol} → {e}")
+
+    return None
+
+
+def main():
+    symbols = [
+        f"{b}/{q}"
+        for b in BASE_ASSETS
+        for q in QUOTE_ASSETS
+        if b != q
+    ]
+
+    print(f"Checking {len(symbols)} symbols...\n")
+
+    results = []
+
+    with ThreadPoolExecutor(max_workers=4) as executor:
+        futures = {executor.submit(fetch, s): s for s in symbols}
+
+        for future in as_completed(futures):
+            symbol = futures[future]
+            result = future.result()
+
+            if result:
+                print(f"✔ {symbol}")
+                results.append(result)
+            else:
+                print(f"✘ {symbol}")
+
+    output = {
+        "generated_at": datetime.now(UTC).isoformat(),
+        "total_checked": len(symbols),
+        "total_valid": len(results),
+        "pairs": sorted(results, key=lambda x: x["symbol"])
+    }
+
+    with open(OUTPUT_FILE, "w") as f:
+        json.dump(output, f, indent=2)
+
+    print("\n=== DONE ===")
+    print(f"Valid pairs: {len(results)}")
+    print(f"Saved to: {OUTPUT_FILE}")
+
+
+if __name__ == "__main__":
+    main()

docker-compose.yml

@@ -0,0 +1,7 @@
+services:
+  metals-mcp:
+    build: .
+    ports:
+      - "8515:8515"
+    environment:
+      - METALS_PORT=8515

killserver.sh

@@ -0,0 +1,19 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+PORT="${METALS_PORT:-8515}"
+PIDS="$(lsof -ti tcp:"$PORT" || true)"
+
+if [ -z "$PIDS" ]; then
+  echo "No listeners found on port $PORT"
+  exit 0
+fi
+
+echo "Stopping listeners on port $PORT: $PIDS"
+kill $PIDS || true
+sleep 1
+PIDS="$(lsof -ti tcp:"$PORT" || true)"
+if [ -n "$PIDS" ]; then
+  echo "Force-killing remaining listeners on port $PORT: $PIDS"
+  kill -9 $PIDS || true
+fi

main.py

@@ -0,0 +1,11 @@
+from __future__ import annotations
+
+import uvicorn
+
+from src.metals_mcp.config import HOST, PORT
+from src.metals_mcp.server_fastmcp import create_app
+
+app = create_app()
+
+if __name__ == "__main__":
+    uvicorn.run("main:app", host=HOST, port=PORT, reload=True)

metals-mcp_first_idea.md

@@ -0,0 +1,153 @@
+## Swissquote-Based Market Data MCP Server
+
+### Abstract
+
+This paper proposes a lightweight Market Context Protocol (MCP) server that ingests real-time bid/ask quotes from Swissquote’s public forex data feed and exposes normalized price and candle data as reusable tools for downstream applications. The system is designed for macro analysis, market orientation, and local-first architectures where free and consistent data is preferred over institutional-grade feeds.
+
+---
+
+### 1. Motivation
+
+Most free financial APIs provide aggregated or delayed data with limited transparency. Swissquote’s public BBO feed offers a rare combination of:
+
+* Low latency
+* Direct bid/ask quotes
+* Broad instrument coverage (FX + metals)
+
+However, it lacks:
+
+* Historical data
+* Candle aggregation
+* Stable API abstraction
+
+This project addresses these gaps by introducing an MCP-compatible data layer.
+
+---
+
+### 2. System Architecture
+
+**Data Flow:**
+
+```
+Swissquote Feed → Ingestion Layer → Tick Store → Candle Engine → MCP Server → Client Apps
+```
+
+#### Components:
+
+1. **Ingestion Layer**
+
+   * Polls Swissquote endpoints at fixed intervals
+   * Normalizes bid/ask + timestamp
+   * Computes mid-price
+
+2. **Tick Store**
+
+   * Append-only time series storage
+   * Retains raw ticks for replay and recomputation
+   * Suggested: SQLite / TimescaleDB / in-memory ring buffer
+
+3. **Candle Engine**
+
+   * Aggregates ticks into OHLC candles
+   * Uses **clock-aligned 5m candles as the base unit**
+   * Candle windows are fixed, e.g. `00:00–00:05`, `00:05–00:10`
+   * Handles:
+
+     * Missing ticks
+     * Market gaps
+     * Session boundaries
+
+4. **MCP Server**
+
+   * Exposes tools:
+
+     * `get_price(symbol)`
+     * `get_candles(symbol, timeframe, limit)`
+     * `get_last_candle(symbol, timeframe)`
+   * Stateless interface over stateful backend
+
+---
+
+### 3. Data Model
+
+#### Tick:
+
+```json
+{
+  "symbol": "XAU/USD",
+  "bid": 2334.12,
+  "ask": 2334.45,
+  "mid": 2334.285,
+  "timestamp": 1710000000000
+}
+```
+
+#### Candle:
+
+```json
+{
+  "symbol": "XAU/USD",
+  "timeframe": "1m",
+  "open": 2330.10,
+  "high": 2335.00,
+  "low": 2329.80,
+  "close": 2334.20,
+  "start": 1710000000000
+}
+```
+
+---
+
+### 4. Design Principles
+
+* **Determinism over accuracy**
+  Consistent data is more valuable than theoretically perfect data.
+
+* **Clock-aligned candles**
+  A fixed 5-minute window is the primary market unit, even if not every trade is observed.
+
+* **Raw data preservation**
+  Store ticks when useful, but the system can remain simple if 5m candles are the base layer.
+
+* **Separation of concerns**
+  Ingestion, aggregation, and serving are independent layers.
+
+* **Market orientation first**
+  The goal is perception and regime reading, not execution.
+
+* **Stateless interface**
+  MCP tools should not depend on internal state assumptions.
+
+---
+
+### Intended shape
+
+- Primary instruments: gold, silver, and a small relevant subset of metals/forex pairs
+- Primary timeframe: 5m
+- Retention target: 180 days to 1 year, depending on storage needs
+- Use case: understand market structure, world dynamics, and sentiment context
+
+---
+
+### 5. Limitations
+
+* Broker-derived pricing (not exchange-grade)
+* No guaranteed uptime or SLA
+* Possible anomalies during market open/close
+
+---
+
+### 6. Extensions
+
+* Multi-source aggregation (fallback feeds)
+* Volume estimation heuristics
+* Spread analytics
+* Event detection (volatility spikes)
+
+---
+
+### 7. Conclusion
+
+A Swissquote-based MCP server provides a robust and flexible foundation for macro-level analysis and experimental trading systems. While not suitable for high-frequency trading, it offers an excellent balance between cost, control, and data quality for independent developers.
+
+---

requirements.txt

@@ -0,0 +1,6 @@
+fastmcp>=2.0.0
+fastapi>=0.115.0
+uvicorn[standard]>=0.30.0
+pydantic>=2.8.0
+requests>=2.32.0
+pytest>=8.0.0

restart.sh

@@ -0,0 +1,5 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+./killserver.sh
+./run.sh

run.sh

@@ -0,0 +1,18 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+mkdir -p logs
+
+if [ -f .venv/bin/activate ]; then
+  # shellcheck disable=SC1091
+  source .venv/bin/activate
+fi
+
+if [ -f logs/server.pid ] && kill -0 "$(cat logs/server.pid)" 2>/dev/null; then
+  echo "metals-mcp already running on pid $(cat logs/server.pid)"
+  exit 0
+fi
+
+nohup python -m uvicorn main:app --host 0.0.0.0 --port "${METALS_PORT:-8515}" > logs/server.log 2>&1 &
+echo $! > logs/server.pid
+echo "metals-mcp started on pid $(cat logs/server.pid)"

src/metals_mcp/config.py

@@ -0,0 +1,13 @@
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+BASE_DIR = Path(__file__).resolve().parent.parent.parent
+HOST = os.getenv("METALS_HOST", "127.0.0.1")
+PORT = int(os.getenv("METALS_PORT", "8515"))
+DATA_DIR = Path(os.getenv("METALS_DATA_DIR", BASE_DIR / "data"))
+LOG_DIR = Path(os.getenv("METALS_LOG_DIR", BASE_DIR / "logs"))
+DB_PATH = Path(os.getenv("METALS_DB_PATH", DATA_DIR / "metals.sqlite3"))
+POLL_INTERVAL_SECONDS = float(os.getenv("SWISSQUOTE_POLL_INTERVAL_SECONDS", "0.2"))
+METALS_PAIRS = [p.strip().upper() for p in os.getenv("METALS_PAIRS", "XAU/USD,XAG/USD").split(",") if p.strip()]

src/metals_mcp/mcp_tools.py

@@ -0,0 +1,74 @@
+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}

src/metals_mcp/poller.py

@@ -0,0 +1,87 @@
+from __future__ import annotations
+
+import math
+import time
+from dataclasses import dataclass
+from typing import Any
+import logging
+
+from .config import DB_PATH, METALS_PAIRS, POLL_INTERVAL_SECONDS
+from .storage import init_db, upsert_candle
+from .swissquote import SwissquoteClient
+
+TIMEFRAME_SECONDS = 300
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class CandleState:
+    symbol: str
+    timeframe: str
+    start_ts: int
+    open: float
+    high: float
+    low: float
+    close: float
+
+    def update(self, price: float) -> None:
+        self.high = max(self.high, price)
+        self.low = min(self.low, price)
+        self.close = price
+
+    def to_row(self) -> dict[str, Any]:
+        return {
+            "symbol": self.symbol,
+            "timeframe": self.timeframe,
+            "open": self.open,
+            "high": self.high,
+            "low": self.low,
+            "close": self.close,
+            "start_ts": self.start_ts,
+            "end_ts": self.start_ts + TIMEFRAME_SECONDS * 1000,
+        }
+
+
+class CandlePoller:
+    def __init__(self) -> None:
+        self.client = SwissquoteClient()
+        self.states: dict[str, CandleState] = {}
+        init_db(DB_PATH)
+
+    def bucket_start(self, ts_ms: int) -> int:
+        return (ts_ms // (TIMEFRAME_SECONDS * 1000)) * (TIMEFRAME_SECONDS * 1000)
+
+    def step(self) -> None:
+        for symbol in METALS_PAIRS:
+            quote = self.client.fetch_quote(symbol)
+            if not quote:
+                continue
+            start_ts = self.bucket_start(quote.timestamp)
+            state = self.states.get(symbol)
+            if state is None or state.start_ts != start_ts:
+                if state is not None:
+                    upsert_candle(DB_PATH, state.to_row())
+                self.states[symbol] = CandleState(
+                    symbol=symbol,
+                    timeframe="5m",
+                    start_ts=start_ts,
+                    open=quote.mid,
+                    high=quote.mid,
+                    low=quote.mid,
+                    close=quote.mid,
+                )
+            else:
+                state.update(quote.mid)
+
+    def flush(self) -> None:
+        for state in self.states.values():
+            upsert_candle(DB_PATH, state.to_row())
+
+    def run_forever(self) -> None:
+        init_db(DB_PATH)
+        while True:
+            try:
+                self.step()
+            except Exception as exc:
+                logger.exception("metals poller cycle failed: %s", exc)
+            time.sleep(POLL_INTERVAL_SECONDS)

src/metals_mcp/server_fastmcp.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+from fastapi import FastAPI
+from mcp.server.fastmcp import FastMCP
+from mcp.server.transport_security import TransportSecuritySettings
+from threading import Thread
+
+from .config import DB_PATH
+from .mcp_tools import (
+    get_capabilities as _get_capabilities,
+    get_health,
+    get_indicator as _get_indicator,
+    get_market_snapshot as _get_market_snapshot,
+    get_ohlcv as _get_ohlcv,
+    get_price as _get_price,
+    get_regime as _get_regime,
+    get_top_movers as _get_top_movers,
+)
+from .poller import CandlePoller
+from .storage import init_db
+
+mcp = FastMCP(
+    "metals-mcp",
+    transport_security=TransportSecuritySettings(
+        enable_dns_rebinding_protection=False,
+    ),
+)
+app = FastAPI(title="metals-mcp")
+
+
+@mcp.tool()
+def get_price(symbol: str):
+    return _get_price(symbol)
+
+
+@mcp.tool()
+def get_ohlcv(symbol: str, timeframe: str = "5m", limit: int = 100):
+    return _get_ohlcv(symbol, timeframe=timeframe, limit=limit)
+
+
+@mcp.tool()
+def get_indicator(symbol: str, indicator: str, timeframe: str = "5m", params: dict | None = None):
+    return _get_indicator(symbol, indicator, timeframe=timeframe, params=params)
+
+
+@mcp.tool()
+def get_market_snapshot(symbol: str):
+    return _get_market_snapshot(symbol)
+
+
+@mcp.tool()
+def get_top_movers(limit: int = 10):
+    return _get_top_movers(limit=limit)
+
+
+@mcp.tool()
+def get_capabilities():
+    return _get_capabilities()
+
+
+@mcp.tool()
+def get_regime(symbol: str, timeframe: str = "5m"):
+    return _get_regime(symbol, timeframe=timeframe)
+
+
+@app.get("/")
+def root():
+    return {"ok": True, "server": "metals-mcp", "tools": ["get_price", "get_candles", "get_last_candle", "get_capabilities"]}
+
+
+@app.get("/health")
+def health():
+    return get_health()
+
+
+app.mount("/mcp", mcp.sse_app())
+_poller_started = False
+
+
+def _start_poller() -> None:
+    global _poller_started
+    if _poller_started:
+        return
+    init_db(DB_PATH)
+    Thread(target=CandlePoller().run_forever, daemon=True, name="metals-poller").start()
+    _poller_started = True
+
+
+@app.on_event("startup")
+def startup() -> None:
+    _start_poller()
+
+
+def create_app() -> FastAPI:
+    _start_poller()
+    return app

src/metals_mcp/storage.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+import sqlite3
+from pathlib import Path
+from typing import Any
+
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS candles (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    symbol TEXT NOT NULL,
+    timeframe TEXT NOT NULL,
+    open REAL NOT NULL,
+    high REAL NOT NULL,
+    low REAL NOT NULL,
+    close REAL NOT NULL,
+    start_ts INTEGER NOT NULL,
+    end_ts INTEGER NOT NULL,
+    UNIQUE(symbol, timeframe, start_ts)
+);
+"""
+
+
+def connect(db_path: str | Path) -> sqlite3.Connection:
+    conn = sqlite3.connect(str(db_path))
+    conn.row_factory = sqlite3.Row
+    return conn
+
+
+def init_db(db_path: str | Path) -> None:
+    path = Path(db_path)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with connect(path) as conn:
+        conn.executescript(SCHEMA)
+        conn.commit()
+
+
+def upsert_candle(db_path: str | Path, candle: dict[str, Any]) -> None:
+    with connect(db_path) as conn:
+        conn.execute(
+            """
+            INSERT INTO candles(symbol, timeframe, open, high, low, close, start_ts, end_ts)
+            VALUES(?, ?, ?, ?, ?, ?, ?, ?)
+            ON CONFLICT(symbol, timeframe, start_ts) DO UPDATE SET
+                open=excluded.open,
+                high=excluded.high,
+                low=excluded.low,
+                close=excluded.close,
+                end_ts=excluded.end_ts
+            """,
+            (
+                candle["symbol"],
+                candle["timeframe"],
+                candle["open"],
+                candle["high"],
+                candle["low"],
+                candle["close"],
+                candle["start_ts"],
+                candle["end_ts"],
+            ),
+        )
+        conn.commit()
+
+
+def latest_candles(db_path: str | Path, symbol: str, timeframe: str, limit: int = 100) -> list[dict[str, Any]]:
+    with connect(db_path) as conn:
+        rows = conn.execute(
+            """
+            SELECT symbol, timeframe, open, high, low, close, start_ts, end_ts
+            FROM candles
+            WHERE symbol = ? AND timeframe = ?
+            ORDER BY start_ts DESC
+            LIMIT ?
+            """,
+            (symbol, timeframe, limit),
+        ).fetchall()
+    return [dict(row) for row in reversed(rows)]
+
+
+def last_candle(db_path: str | Path, symbol: str, timeframe: str) -> dict[str, Any] | None:
+    with connect(db_path) as conn:
+        row = conn.execute(
+            """
+            SELECT symbol, timeframe, open, high, low, close, start_ts, end_ts
+            FROM candles
+            WHERE symbol = ? AND timeframe = ?
+            ORDER BY start_ts DESC
+            LIMIT 1
+            """,
+            (symbol, timeframe),
+        ).fetchone()
+    return dict(row) if row else None
+
+
+def stats(db_path: str | Path) -> dict[str, Any]:
+    with connect(db_path) as conn:
+        candles = conn.execute("SELECT COUNT(*) AS n FROM candles").fetchone()["n"]
+    return {"candles": candles}

src/metals_mcp/swissquote.py

@@ -0,0 +1,59 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from typing import Any
+
+import requests
+
+BASE_URL = "https://forex-data-feed.swissquote.com/public-quotes/bboquotes/instrument/{}"
+HEADERS = {
+    "User-Agent": "Mozilla/5.0",
+    "Accept": "application/json, text/plain, */*",
+}
+
+
+@dataclass(frozen=True)
+class Quote:
+    symbol: str
+    bid: float
+    ask: float
+    timestamp: int
+
+    @property
+    def mid(self) -> float:
+        return (self.bid + self.ask) / 2.0
+
+
+class SwissquoteClient:
+    def normalize_symbol(self, symbol: str) -> str:
+        cleaned = symbol.replace("/", "").upper()
+        if cleaned in {"XAU", "XAG", "XPT", "XPD"}:
+            return f"{cleaned}/USD"
+        if "/" in symbol:
+            return symbol.upper()
+        return symbol.upper()
+
+    def fetch_quote(self, symbol: str) -> Quote | None:
+        normalized = self.normalize_symbol(symbol)
+        response = requests.get(BASE_URL.format(normalized), headers=HEADERS, timeout=5)
+        response.raise_for_status()
+        data: Any = response.json()
+        items = data if isinstance(data, list) else [data]
+        for item in items:
+            if not isinstance(item, dict):
+                continue
+            prices = item.get("spreadProfilePrices") or []
+            if not prices:
+                continue
+            price = prices[0]
+            bid = float(price.get("bid", 0))
+            ask = float(price.get("ask", 0))
+            ts = item.get("ts") or price.get("ts") or price.get("timestamp")
+            if bid > 0 and ask > 0 and ts:
+                return Quote(symbol=symbol, bid=bid, ask=ask, timestamp=int(ts))
+        return None
+
+
+def now_ms() -> int:
+    return int(datetime.now(tz=timezone.utc).timestamp() * 1000)

swissquote_pairs.json

@@ -0,0 +1,175 @@
+{
+  "generated_at": "2026-04-11T07:52:18.355355+00:00",
+  "total_checked": 55,
+  "total_valid": 28,
+  "pairs": [
+    {
+      "symbol": "AUD/CHF",
+      "bid": 0.55798,
+      "ask": 0.55823,
+      "timestamp": null
+    },
+    {
+      "symbol": "AUD/JPY",
+      "bid": 112.639,
+      "ask": 112.675,
+      "timestamp": null
+    },
+    {
+      "symbol": "AUD/USD",
+      "bid": 0.70717,
+      "ask": 0.70734,
+      "timestamp": null
+    },
+    {
+      "symbol": "CAD/CHF",
+      "bid": 0.56995,
+      "ask": 0.57026,
+      "timestamp": null
+    },
+    {
+      "symbol": "CAD/JPY",
+      "bid": 115.07,
+      "ask": 115.106,
+      "timestamp": null
+    },
+    {
+      "symbol": "CHF/JPY",
+      "bid": 201.841,
+      "ask": 201.885,
+      "timestamp": null
+    },
+    {
+      "symbol": "EUR/CHF",
+      "bid": 0.92543,
+      "ask": 0.92584,
+      "timestamp": null
+    },
+    {
+      "symbol": "EUR/GBP",
+      "bid": 0.87087,
+      "ask": 0.87105,
+      "timestamp": null
+    },
+    {
+      "symbol": "EUR/JPY",
+      "bid": 186.825,
+      "ask": 186.866,
+      "timestamp": null
+    },
+    {
+      "symbol": "EUR/USD",
+      "bid": 1.17291,
+      "ask": 1.17311,
+      "timestamp": null
+    },
+    {
+      "symbol": "GBP/CHF",
+      "bid": 1.06261,
+      "ask": 1.06297,
+      "timestamp": null
+    },
+    {
+      "symbol": "GBP/JPY",
+      "bid": 214.512,
+      "ask": 214.564,
+      "timestamp": null
+    },
+    {
+      "symbol": "GBP/USD",
+      "bid": 1.34672,
+      "ask": 1.34695,
+      "timestamp": null
+    },
+    {
+      "symbol": "NZD/CHF",
+      "bid": 0.46076,
+      "ask": 0.46108,
+      "timestamp": null
+    },
+    {
+      "symbol": "NZD/JPY",
+      "bid": 93.025,
+      "ask": 93.061,
+      "timestamp": null
+    },
+    {
+      "symbol": "NZD/USD",
+      "bid": 0.58398,
+      "ask": 0.58424,
+      "timestamp": null
+    },
+    {
+      "symbol": "USD/CHF",
+      "bid": 0.78895,
+      "ask": 0.78927,
+      "timestamp": null
+    },
+    {
+      "symbol": "USD/JPY",
+      "bid": 159.28,
+      "ask": 159.304,
+      "timestamp": null
+    },
+    {
+      "symbol": "XAG/CHF",
+      "bid": 59.8364,
+      "ask": 60.0787,
+      "timestamp": null
+    },
+    {
+      "symbol": "XAG/EUR",
+      "bid": 64.6513,
+      "ask": 64.7588,
+      "timestamp": null
+    },
+    {
+      "symbol": "XAG/GBP",
+      "bid": 56.2979,
+      "ask": 56.5162,
+      "timestamp": null
+    },
+    {
+      "symbol": "XAG/USD",
+      "bid": 75.85,
+      "ask": 75.95,
+      "timestamp": null
+    },
+    {
+      "symbol": "XAU/CHF",
+      "bid": 3746.763,
+      "ask": 3751.368,
+      "timestamp": null
+    },
+    {
+      "symbol": "XAU/EUR",
+      "bid": 4047.099,
+      "ask": 4050.701,
+      "timestamp": null
+    },
+    {
+      "symbol": "XAU/GBP",
+      "bid": 3524.293,
+      "ask": 3528.208,
+      "timestamp": null
+    },
+    {
+      "symbol": "XAU/USD",
+      "bid": 4748.33,
+      "ask": 4750.46,
+      "timestamp": null
+    },
+    {
+      "symbol": "XPD/USD",
+      "bid": 1511.251,
+      "ask": 1531.299,
+      "timestamp": null
+    },
+    {
+      "symbol": "XPT/USD",
+      "bid": 2038.532,
+      "ask": 2056.309,
+      "timestamp": null
+    }
+  ]
+}

test_metals.py

@@ -0,0 +1,12 @@
+from src.metals_mcp.mcp_tools import get_capabilities, get_candles, get_last_candle, get_price
+
+
+def test_capabilities():
+    caps = get_capabilities()
+    assert caps["server"] == "metals-mcp"
+
+
+def test_scaffold_tools():
+    assert get_price("XAU/USD")["symbol"] == "XAU/USD"
+    assert get_candles("XAU/USD")["candles"] == []
+    assert get_last_candle("XAU/USD")["candle"] is None

test_swissquote_api.py

@@ -0,0 +1,7 @@
+import requests
+
+url = "https://forex-data-feed.swissquote.com/public-quotes/bboquotes/instrument/XAU/USD"
+
+r = requests.get(url, headers={"User-Agent": "Mozilla/5.0"})
+print(r.status_code)
+print(r.text)

tests.sh

@@ -0,0 +1,9 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+if [ -f .venv/bin/activate ]; then
+  # shellcheck disable=SC1091
+  source .venv/bin/activate
+fi
+
+python -m pytest -q