backend/mcp_servers/fred_mcp.py

"""FRED (Federal Reserve Economic Data) MCP Server.

This MCP server provides macroeconomic data from the Federal Reserve Economic Data API.
Free tier: Unlimited requests with API key.
"""

import logging
from datetime import datetime
from typing import Dict, List, Optional
from decimal import Decimal

import httpx
from fastmcp import FastMCP
from pydantic import BaseModel, Field
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
)

from backend.config import settings

logger = logging.getLogger(__name__)

# Initialize MCP server
mcp = FastMCP("fred")

# API Configuration
BASE_URL = "https://api.stlouisfed.org/fred"
API_KEY = settings.fred_api_key


class SeriesRequest(BaseModel):
    """Request for economic data series."""

    series_id: str = Field(..., description="FRED series ID (e.g., GDP, UNRATE)")
    observation_start: Optional[str] = Field(None, description="Start date (YYYY-MM-DD)")
    observation_end: Optional[str] = Field(None, description="End date (YYYY-MM-DD)")


class SeriesObservation(BaseModel):
    """Single observation from a FRED series."""

    date: str
    value: Optional[Decimal] = None


class EconomicSeries(BaseModel):
    """Economic data series."""

    series_id: str
    title: Optional[str] = None
    units: Optional[str] = None
    frequency: Optional[str] = None
    observations: List[SeriesObservation] = Field(default_factory=list)


async def _make_request(endpoint: str, params: Dict[str, str]) -> Dict:
    """Make HTTP request to FRED API."""
    params["api_key"] = API_KEY
    params["file_type"] = "json"

    url = f"{BASE_URL}/{endpoint}"

    async with httpx.AsyncClient() as client:
        response = await client.get(url, params=params, timeout=30.0)
        response.raise_for_status()
        return response.json()


@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type((TimeoutError, ConnectionError, Exception)),
)
@mcp.tool()
async def get_economic_series(request: SeriesRequest) -> EconomicSeries:
    """Get economic data series from FRED.

    Args:
        request: Series request

    Returns:
        Economic series with observations

    Example:
        >>> await get_economic_series(SeriesRequest(series_id="GDP"))

    Common Series IDs:
        - GDP: Gross Domestic Product
        - UNRATE: Unemployment Rate
        - CPIAUCSL: Consumer Price Index
        - DFF: Federal Funds Rate
        - T10Y2Y: 10-Year minus 2-Year Treasury Yield
    """
    logger.info(f"Fetching FRED series {request.series_id}")

    try:
        # Get series info
        series_params = {"series_id": request.series_id}
        series_data = await _make_request("series", series_params)
        series_info = series_data.get("seriess", [{}])[0] if series_data.get("seriess") else {}

        # Get observations
        obs_params = {"series_id": request.series_id}
        if request.observation_start:
            obs_params["observation_start"] = request.observation_start
        if request.observation_end:
            obs_params["observation_end"] = request.observation_end

        obs_data = await _make_request("series/observations", obs_params)
        observations = []

        for obs in obs_data.get("observations", []):
            value = None
            if obs.get("value") and obs["value"] != ".":
                try:
                    value = Decimal(obs["value"])
                except:
                    pass

            observations.append(
                SeriesObservation(date=obs.get("date", ""), value=value)
            )

        series = EconomicSeries(
            series_id=request.series_id,
            title=series_info.get("title"),
            units=series_info.get("units"),
            frequency=series_info.get("frequency"),
            observations=observations,
        )

        logger.info(f"Fetched {len(observations)} observations for {request.series_id}")
        return series

    except Exception as e:
        logger.error(f"Error fetching FRED series {request.series_id}: {e}")
        return EconomicSeries(series_id=request.series_id)


# Export the MCP server
if __name__ == "__main__":
    mcp.run()