fetch_cpis_data
Retrieve CPIS database time series data for specified countries, indicators, and date ranges to analyze international investment positions.
Instructions
Retrieves compact format time series data from the CPIS database based on the input parameters.
Args:
freq (str): Frequency (e.g., "A" for annual).
country (str): Country code, multiple country codes can be connected with "+".
indicator (str): Indicator code.
counter_country (str): Counterpart country code.
start (str | int): Start year.
end (str | int): End year.
Returns:
str: Description of the queried data. Do not perform further analysis or retry if the query fails.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| freq | Yes | ||
| country | Yes | ||
| indicator | Yes | ||
| counter_country | Yes | ||
| start | Yes | ||
| end | Yes |
Implementation Reference
- imf_data_mcp/__init__.py:175-200 (handler)The main handler function for the 'fetch_cpis_data' tool, registered via @mcp.tool(). It constructs the IMF CPIS API URL based on parameters, fetches JSON data, processes it with process_imf_data, and returns a string summary or error.@mcp.tool() def fetch_cpis_data(freq: str, country: str, indicator: str, counter_country: str, start: str | int, end: str | int) -> str: """ Retrieves compact format time series data from the CPIS database based on the input parameters. Args: freq (str): Frequency (e.g., "A" for annual). country (str): Country code, multiple country codes can be connected with "+". indicator (str): Indicator code. counter_country (str): Counterpart country code. start (str | int): Start year. end (str | int): End year. Returns: str: Description of the queried data. Do not perform further analysis or retry if the query fails. """ dimensions = f"{freq}.{country}.{indicator}.T.T.{counter_country}" url = f"http://dataservices.imf.org/REST/SDMX_JSON.svc/CompactData/CPIS/{dimensions}?startPeriod={start}&endPeriod={end}" try: response = requests.get(url) response.raise_for_status() data = response.json() return process_imf_data(data) except Exception as e: return f"Error fetching CPIS data: {str(e)}"
- imf_data_mcp/utils.py:1-52 (helper)Helper utility function used by fetch_cpis_data (and other fetch tools) to parse IMF CompactData JSON into a human-readable string summary of observations per country and time period.def process_imf_data(json_data: dict) -> str: """ Process IMF data and return a string with the information. :param: json_data(dict): JSON data from the IMF API :return: (str) A string with the information from the JSON data """ try: json_data = json_data["CompactData"] dataset = json_data["DataSet"] series_list = dataset["Series"] if isinstance(series_list, dict): series_list = [series_list] elif not isinstance(series_list, list): return f"Error: Expected series_list to be a list but got {type(series_list)}" output_texts = [] for series in series_list: if series is None: output_texts.append("Warning: No indicator value.") continue country = series.get("@REF_AREA", None) obs = series.get("Obs", {}) if isinstance(obs, dict): obs = [obs] elif not isinstance(obs, list): return f"Error: Expected obs to be a list but got {type(obs)}" for _obs in obs: if _obs is None: output_texts.append( f"Warning: No indicator value for {country} in that Year, You should not try to access the data of this country." ) continue time_period = _obs.get("@TIME_PERIOD", "that Year") obs_value = _obs.get("@OBS_VALUE") if obs_value is not None: text = f"In {time_period}, {country} had an indicator value of {float(obs_value):.2f}." output_texts.append(text) else: output_texts.append(f"Warning: No indicator value for {country} in {time_period}.") return "\n".join(output_texts) except KeyError as e: return f"Error processing IMF data: Missing key {str(e)}" except Exception as e: return f"Error processing IMF data: {str(e)}"