fetch_dot_data
Retrieve compact time series data from the DOT database by specifying frequency, country, indicator, counterpart, and date range. Supports IMF Data API integration for streamlined data access.
Instructions
Retrieves compact format time series data from the DOT 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.
counterpart (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 |
|---|---|---|---|
| counterpart | Yes | ||
| country | Yes | ||
| end | Yes | ||
| freq | Yes | ||
| indicator | Yes | ||
| start | Yes |
Implementation Reference
- imf_data_mcp/__init__.py:95-120 (handler)The handler function for the fetch_dot_data tool. It is registered via @mcp.tool() decorator. Fetches time series data from IMF DOT (Direction of Trade Statistics) dataset using CompactData API, processes the JSON response with process_imf_data helper, and returns a textual summary.@mcp.tool() def fetch_dot_data(freq: str, country: str, indicator: str, counterpart: str, start: str | int, end: str | int) -> str: """ Retrieves compact format time series data from the DOT 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. counterpart (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}.{counterpart}" url = f"http://dataservices.imf.org/REST/SDMX_JSON.svc/CompactData/DOT/{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 DOT data: {str(e)}"
- imf_data_mcp/utils.py:1-51 (helper)Helper utility function that processes the raw JSON data from IMF CompactData API into a human-readable string format listing observations by 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)}"