BLS MCP Server

Instructions

Implementation Reference

• src/bls_mcp/tools/get_series.py:51-89 (handler)

  The `execute` method in `GetSeriesTool` class implements the core logic of the "get_series" tool: input validation using schema and helpers, data fetching from provider, error handling, and returning results.

  async def execute(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
      """Execute get_series tool."""
      logger.info(f"Executing get_series with arguments: {arguments}")
  
      # Validate input
      try:
          input_data = GetSeriesInput(**arguments)
      except Exception as e:
          logger.error(f"Input validation failed: {e}")
          return {"error": f"Invalid input: {str(e)}"}
  
      # Validate series ID format
      if not validate_series_id(input_data.series_id):
          return {"error": f"Invalid series ID format: {input_data.series_id}"}
  
      # Validate year range
      is_valid, error_msg = validate_year_range(
          input_data.start_year, input_data.end_year
      )
      if not is_valid:
          return {"error": error_msg}
  
      # Fetch data
      try:
          result = await self.data_provider.get_series(
              series_id=input_data.series_id,
              start_year=input_data.start_year,
              end_year=input_data.end_year,
          )
          logger.info(
              f"Successfully fetched {result['count']} data points for {input_data.series_id}"
          )
          return result
      except ValueError as e:
          logger.warning(f"Series not found: {e}")
          return {"error": str(e)}
      except Exception as e:
          logger.error(f"Error fetching series: {e}")
          return {"error": f"Failed to fetch series: {str(e)}"}

• src/bls_mcp/tools/get_series.py:15-27 (schema)

  Pydantic `BaseModel` defining the input schema for the get_series tool, including series_id, optional start_year and end_year with descriptions.

  class GetSeriesInput(BaseModel):
      """Input schema for get_series tool."""
  
      series_id: str = Field(
          description="BLS series ID (e.g., 'CUUR0000SA0' for CPI All Items)"
      )
      start_year: Optional[int] = Field(
          default=None, description="Start year for data range (optional)"
      )
      end_year: Optional[int] = Field(
          default=None, description="End year for data range (optional)"
      )

• src/bls_mcp/server.py:47-52 (registration)

  Registration of the GetSeriesTool instance in the server's tools dictionary, which is used by list_tools and call_tool handlers.

  self.tools = {
      "get_series": GetSeriesTool(self.data_provider),
      "list_series": ListSeriesTool(self.data_provider),
      "get_series_info": GetSeriesInfoTool(self.data_provider),
      "plot_series": PlotSeriesTool(self.data_provider),
  }

• src/bls_mcp/data/mock_data.py:33-84 (helper)

  The `get_series` method in `MockDataProvider` that loads mock data from fixtures, filters by year range if provided, attaches metadata, and returns the series data. Called by the tool handler.

  async def get_series(
      self,
      series_id: str,
      start_year: int | None = None,
      end_year: int | None = None,
  ) -> dict[str, Any]:
      """
      Get data for a specific series.
  
      Args:
          series_id: BLS series ID (e.g., 'CUUR0000SA0')
          start_year: Optional start year filter
          end_year: Optional end year filter
  
      Returns:
          Dictionary with series data
  
      Raises:
          ValueError: If series not found
      """
      historical = self._load_historical_data()
  
      if series_id not in historical:
          raise ValueError(f"Series '{series_id}' not found in mock data")
  
      series_data = historical[series_id]
      data_points = series_data["data"]
  
      # Filter by year range if specified
      if start_year is not None or end_year is not None:
          filtered_points = []
          for point in data_points:
              year = int(point["year"])
              if start_year and year < start_year:
                  continue
              if end_year and year > end_year:
                  continue
              filtered_points.append(point)
          data_points = filtered_points
  
      # Get series metadata
      catalog = self._load_series_catalog()
      metadata: dict[str, Any] = next(
          (s for s in catalog["series"] if s["series_id"] == series_id), {}
      )
  
      return {
          "series_id": series_id,
          "data": data_points,
          "metadata": metadata,
          "count": len(data_points),
      }