FrankfurterMCP

Instructions

Implementation Reference

• src/frankfurtermcp/server.py:253-322 (handler)

  The main async handler function `get_historical_exchange_rates` that processes tool requests. It accepts parameters for base_currency, symbols, specific_date, start_date, and end_date (all validated with Pydantic Field annotations). It calls the helper `_get_historical_exchange_rates` and returns results via `get_response_content`.

  async def get_historical_exchange_rates(
      self,
      ctx: Context,
      base_currency: Annotated[
          ISO4217,
          Field(description="A base currency ISO4217 code for which rates are to be requested."),
      ],
      symbols: Annotated[
          list[ISO4217] | ISO4217 | None,
          Field(
              description="A list of target currency ISO4217 codes for which rates against the base currency will be provided. If not provided, all supported currencies will be shown."
          ),
      ] = None,
      specific_date: Annotated[
          date | None,
          Field(
              default=None,
              description="The specific date for which the historical rates are requested in the YYYY-MM-DD format.",
          ),
      ] = None,
      start_date: Annotated[
          date | None,
          Field(
              default=None,
              description="The start date, of a date range, for which the historical rates are requested in the YYYY-MM-DD format.",
          ),
      ] = None,
      end_date: Annotated[
          date | None,
          Field(
              default=None,
              description="The end date, of a date range, for which the historical rates are requested in the YYYY-MM-DD format.",
          ),
      ] = None,
  ):
      """Returns historical exchange rates for a specific date or date range.
  
      If the exchange rates for a specified date is not available, the rates available for
      the closest date before the specified date will be provided.
      Either a specific date, a start date, or a date range must be provided.
      The symbols can be used to filter the results to specific currencies.
      If symbols are not provided, all supported currencies will be returned.
      """
      await ctx.info(f"Fetching historical exchange rates from Frankfurter API at {self.frankfurter_api_url}")
      # Some LLMs make this mistake of passing just one currency but not as a list!
      if type(symbols) is str:
          symbols = [symbols]
      cache_key = hashkey(
          self,
          specific_date=specific_date.isoformat() if specific_date else None,
          start_date=start_date.isoformat() if start_date else None,
          end_date=end_date.isoformat() if end_date else None,
          base_currency=base_currency,
          symbols=tuple(symbols) if symbols else None,
      )
      cache_hit = cache_key in lru_cache
      result, http_response = self._get_historical_exchange_rates(
          specific_date=specific_date.isoformat() if specific_date else None,
          start_date=start_date.isoformat() if start_date else None,
          end_date=end_date.isoformat() if end_date else None,
          base_currency=base_currency,
          symbols=tuple(symbols) if symbols else None,
      )
      if cache_hit:
          await ctx.info(
              f"Historical exchange rates fetched for {len(result.get('rates', []))} dates from least-recently used (LRU) cache."
          )
      else:
          await ctx.info(f"Historical exchange rates fetched for {len(result.get('rates', []))} dates.")
      return self.get_response_content(response=result, http_response=http_response, cached_response=cache_hit)

• src/frankfurtermcp/server.py:128-166 (helper)

  The private helper function `_get_historical_exchange_rates` that performs the actual HTTP request to the Frankfurter API. It's decorated with `@cached(cache=lru_cache)` for caching and constructs the API URL based on date parameters (specific_date, start_date, end_date) and currency filters.

  @cached(cache=lru_cache, lock=threading.Lock(), key=hashkey)
  def _get_historical_exchange_rates(
      self,
      specific_date: str | None = None,
      start_date: str | None = None,
      end_date: str | None = None,
      base_currency: str | None = None,
      symbols: tuple[str, ...] | None = None,
  ):
      """Internal function to get historical exchange rates. This is a helper function for the main tool."""
      try:
          params = {}
          if base_currency:
              params["base"] = base_currency
          if symbols:
              params["symbols"] = ",".join(symbols)
  
          frankfurter_url = self.frankfurter_api_url
          if start_date and end_date:
              frankfurter_url += f"/{start_date}..{end_date}"
          elif start_date:
              # If only start_date is provided, we assume the end date is the latest available date
              frankfurter_url += f"/{start_date}.."
          elif specific_date:
              # If only specific_date is provided, we assume it is the date for which we want the rates
              frankfurter_url += f"/{specific_date}"
          else:
              raise ValueError("You must provide either a specific date, a start date, or a date range.")
  
          with self.get_httpx_client() as client:
              http_response = client.get(
                  frankfurter_url,
                  params=params,
              )
              http_response.raise_for_status()
              result = http_response.json()
              return result, http_response
      except httpx.RequestError as e:
          raise ValueError(f"Failed to fetch historical exchange rates from {self.frankfurter_api_url}. {e}")

• src/frankfurtermcp/server.py:55-62 (registration)

  The tool registration entry in the `tools` class variable list. It defines 'get_historical_exchange_rates' with tags ['currency-rates', 'historical-exchange-rates'] and annotations indicating it's read-only and open-world.

  {
      "fn": "get_historical_exchange_rates",
      "tags": ["currency-rates", "historical-exchange-rates"],
      "annotations": {
          "readOnlyHint": True,
          "openWorldHint": True,
      },
  },

• src/frankfurtermcp/mixin.py:37-53 (registration)

  The `register_features` method in MCPMixin that iterates over the `tools` list and registers each tool with the FastMCP instance using `mcp.tool()`. This is where the tool function name ('fn' key) is resolved to the actual method and registered.

  def register_features(self, mcp: FastMCP) -> FastMCP:
      """Register tools, resources, and prompts with the given FastMCP instance.
  
      Args:
          mcp (FastMCP): The FastMCP instance to register features with.
  
      Returns:
          FastMCP: The FastMCP instance with registered features.
      """
      # Register tools
      for tool in self.tools:
          assert "fn" in tool, "Tool metadata must include the 'fn' key."
          tool_copy = copy.deepcopy(tool)
          fn_name = tool_copy.pop("fn")
          fn = getattr(self, fn_name)
          mcp.tool(**tool_copy)(fn)
          logger.debug(f"Registered MCP tool: {fn_name}")

• src/frankfurtermcp/server.py:256-286 (schema)

  The input schema/parameter definitions for the tool using Pydantic's Annotated types with Field descriptors. Defines validated parameters: base_currency (ISO4217), symbols (list of ISO4217 or single), specific_date, start_date, and end_date (all date types with descriptions).

  base_currency: Annotated[
      ISO4217,
      Field(description="A base currency ISO4217 code for which rates are to be requested."),
  ],
  symbols: Annotated[
      list[ISO4217] | ISO4217 | None,
      Field(
          description="A list of target currency ISO4217 codes for which rates against the base currency will be provided. If not provided, all supported currencies will be shown."
      ),
  ] = None,
  specific_date: Annotated[
      date | None,
      Field(
          default=None,
          description="The specific date for which the historical rates are requested in the YYYY-MM-DD format.",
      ),
  ] = None,
  start_date: Annotated[
      date | None,
      Field(
          default=None,
          description="The start date, of a date range, for which the historical rates are requested in the YYYY-MM-DD format.",
      ),
  ] = None,
  end_date: Annotated[
      date | None,
      Field(
          default=None,
          description="The end date, of a date range, for which the historical rates are requested in the YYYY-MM-DD format.",
      ),
  ] = None,