get_series
Fetch Bureau of Labor Statistics time series data by ID with optional date range filtering to retrieve economic indicators like CPI and employment statistics.
Instructions
Fetch BLS data series by ID with optional date range filtering. Returns time series data points with values, periods, and metadata.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| series_id | Yes | BLS series ID (e.g., 'CUUR0000SA0' for CPI All Items) | |
| start_year | No | Start year for data range (optional) | |
| end_year | No | End year for data range (optional) |
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)}"}
- 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), }