MCP-FREDAPI

Instructions

Implementation Reference

• server.py:31-93 (handler)

  The main handler function implementing the tool logic: constructs API parameters from inputs, calls the make_request helper to fetch data from FRED API /series/observations endpoint, extracts observations, and returns them.
  async def get_fred_series_observations( series_id: Annotated[str, Field(description="The id for a series.")], realtime_start: Annotated[Optional[str], Field(description="The start of the real-time period. Format: YYYY-MM-DD. Defaults to today's date.")] = None, realtime_end: Annotated[Optional[str], Field(description="The end of the real-time period. Format: YYYY-MM-DD. Defaults to today's date.")] = None, limit: Annotated[Optional[int | str], Field(description="Maximum number of observations to return. Defaults to 10.")] = 10, offset: Annotated[Optional[int | str], Field(description="Number of observations to offset from first. Defaults to 0.")] = 0, sort_order: Annotated[Literal['asc', 'desc'], Field(description="Sort order of observations. Options: 'asc' or 'desc'. Defaults to 'asc'.")] = 'asc', observation_start: Annotated[Optional[str], Field(description="Start date of observations. Format: YYYY-MM-DD.")] = None, observation_end: Annotated[Optional[str], Field(description="End date of observations. Format: YYYY-MM-DD.")] = None, units: Annotated[Literal['lin', 'chg', 'ch1', 'pch', 'pc1', 'pca', 'cch', 'cca', 'log'], Field(description="Data value transformation. Options: 'lin', 'chg', 'ch1', 'pch', 'pc1', 'pca', 'cch', 'cca', 'log'. Defaults to 'lin'.")] = 'lin', frequency: Annotated[Literal['d', 'w', 'bw', 'm', 'q', 'sa', 'a', 'wef', 'weth', 'wew', 'wetu', 'wem', 'wesu', 'wesa', 'bwew', 'bwem'], Field(description="Frequency of observations. Options: 'd', 'w', 'bw', 'm', 'q', 'sa', 'a', 'wef', 'weth', 'wew', 'wetu', 'wem', 'wesu', 'wesa', 'bwew', 'bwem'. Defaults to no value for no frequency aggregation.")] = None, aggregation_method: Annotated[Literal['avg', 'sum', 'eop'], Field(description="Aggregation method for frequency. Options: 'avg', 'sum', 'eop'. Defaults to 'avg'.")] = 'avg', output_type: Annotated[Literal[1, 2, 3, 4], Field(description="Output type of observations. Options: 1, 2, 3, 4. Defaults to 1.")] = 1, vintage_dates: Annotated[Optional[str], Field(description="Comma-separated list of vintage dates.")] = None, ): """Get series observations from the Fred API. Args: series_id (str): The id for a series. realtime_start (str): The start of the real-time period. YYYY-MM-DD formatted string, optional, default: today's date. realtime_end (str): The end of the real-time period. YYYY-MM-DD formatted string, optional, default: today's date. limit (int or str): The maximum number of observations to return. Optional, default: 1000. offset (int or str): The number of observations to offset from the first observation. Optional, default: 0. sort_order (str): The sort order of the observations. Possible values: "asc" or "desc". Optional, default: "asc". observation_start (str): The start date of the observations to get. YYYY-MM-DD formatted string. Optional, default: 1776-07-04 (earliest available). observation_end (str): The end date of the observations to get. YYYY-MM-DD formatted string. Optional, default: 9999-12-31 (latest available). units (str): A key that indicates a data value transformation. Posible values: 'lin', 'chg', 'ch1', 'pch', 'pc1', 'pca', 'cch', 'cca', 'log'. Optional, default: 'lin' (No transformation). frequency (str): The frequency of the observations. Posible values: 'd', 'w', 'bw', 'm', 'q', 'sa', 'a', 'wef', 'weth', 'wew', 'wetu', 'wem', 'wesu', 'wesa', 'bwew', 'bwem'. Optional, default: no value for no frequency aggregation. aggregation_method (str): A key that indicates the aggregation method used for frequency aggregation. This parameter has no affect if the frequency parameter is not set. Posible values: 'avg', 'sum', 'eop'. Optional, default: "avg". output_type (int): The output type of the observations. Optional, default: 1. vintage_dates (str): A comma-separated list of vintage dates to return. Optional, default: no vintage dates are set by default. Returns: dict[str, str]: A dictionary containing the observations or data values for an economic data series. """ params = { "series_id": series_id, # ✅ "realtime_start": realtime_start, # ❌ "realtime_end": realtime_end, # ❌ "limit": limit, # ✅ "offset": offset, # ✅ "sort_order": sort_order, # ✅ "observation_start": observation_start, # ❌ "observation_end": observation_end, # ❌ "units": units, # ✅ "frequency": frequency, # ✅ "aggregation_method": aggregation_method, # ✅ "output_type": output_type, # ✅ "vintage_dates": vintage_dates, # ❌ "file_type": "json" } data = await make_request(f"{FRED_API_URL}/series/observations", params) if not data: raise ConnectionError("Failed to fetch data from the FRED API") observations = data["observations"] if not observations: raise ValueError("No observations found for the given series") return observations
• server.py:31-45 (schema)

  Input schema defined via Annotated types and pydantic Field descriptions, including all parameters for the FRED API /series/observations endpoint.
  async def get_fred_series_observations( series_id: Annotated[str, Field(description="The id for a series.")], realtime_start: Annotated[Optional[str], Field(description="The start of the real-time period. Format: YYYY-MM-DD. Defaults to today's date.")] = None, realtime_end: Annotated[Optional[str], Field(description="The end of the real-time period. Format: YYYY-MM-DD. Defaults to today's date.")] = None, limit: Annotated[Optional[int | str], Field(description="Maximum number of observations to return. Defaults to 10.")] = 10, offset: Annotated[Optional[int | str], Field(description="Number of observations to offset from first. Defaults to 0.")] = 0, sort_order: Annotated[Literal['asc', 'desc'], Field(description="Sort order of observations. Options: 'asc' or 'desc'. Defaults to 'asc'.")] = 'asc', observation_start: Annotated[Optional[str], Field(description="Start date of observations. Format: YYYY-MM-DD.")] = None, observation_end: Annotated[Optional[str], Field(description="End date of observations. Format: YYYY-MM-DD.")] = None, units: Annotated[Literal['lin', 'chg', 'ch1', 'pch', 'pc1', 'pca', 'cch', 'cca', 'log'], Field(description="Data value transformation. Options: 'lin', 'chg', 'ch1', 'pch', 'pc1', 'pca', 'cch', 'cca', 'log'. Defaults to 'lin'.")] = 'lin', frequency: Annotated[Literal['d', 'w', 'bw', 'm', 'q', 'sa', 'a', 'wef', 'weth', 'wew', 'wetu', 'wem', 'wesu', 'wesa', 'bwew', 'bwem'], Field(description="Frequency of observations. Options: 'd', 'w', 'bw', 'm', 'q', 'sa', 'a', 'wef', 'weth', 'wew', 'wetu', 'wem', 'wesu', 'wesa', 'bwew', 'bwem'. Defaults to no value for no frequency aggregation.")] = None, aggregation_method: Annotated[Literal['avg', 'sum', 'eop'], Field(description="Aggregation method for frequency. Options: 'avg', 'sum', 'eop'. Defaults to 'avg'.")] = 'avg', output_type: Annotated[Literal[1, 2, 3, 4], Field(description="Output type of observations. Options: 1, 2, 3, 4. Defaults to 1.")] = 1, vintage_dates: Annotated[Optional[str], Field(description="Comma-separated list of vintage dates.")] = None, ):
• server.py:30-30 (registration)

  MCP tool registration decorator specifying the tool name and description.
  @mcp.tool(name="get_fred_series_observations", description="""Get series observations from the Fred API.""")
• server.py:15-29 (helper)

  Helper function to make authenticated HTTP GET requests to the FRED API, handling API key from env and error raising.
  async def make_request(url: str, params: dict): """Make a request to the Federal Reserve Economic Data API.""" if FRED_API_KEY := os.getenv("FRED_API_KEY"): params["api_key"] = FRED_API_KEY async with httpx.AsyncClient() as client: try: response = await client.get(url, params=params) response.raise_for_status() return response.json() except httpx.HTTPStatusError as e: raise ConnectionError(f"Failed to fetch data from the FRED API: {e}")