FrankfurterMCP

Instructions

Implementation Reference

• src/frankfurtermcp/server.py:324-384 (handler)

  The main handler function that converts an amount from one currency to another using exchange rates for a specific date. It validates currencies, fetches historical exchange rates via the _get_historical_exchange_rates helper, calculates the converted amount, and returns a CurrencyConversionResponse.

  async def convert_currency_specific_date(
      self,
      ctx: Context,
      amount: Annotated[
          PositiveFloat,
          Field(description="The amount in the source currency to convert."),
      ],
      from_currency: Annotated[ISO4217, Field(description="The source currency ISO4217 code.")],
      to_currency: Annotated[ISO4217, Field(description="The target currency ISO4217 code.")],
      specific_date: Annotated[
          date,
          Field(description="The specific date for which the conversion is requested in the YYYY-MM-DD format."),
      ],
  ):
      """Convert an amount from one currency to another using the exchange rates for a specific date.
  
      If there is no exchange rate available for the specific date, the rate for the closest available date before
      the specified date will be used.
      """
      if from_currency.lower() == to_currency.lower():
          # If the source and target currencies are the same, no conversion is needed
          raise ValueError(
              f"Source currency '{from_currency}' and target currency '{to_currency}' are the same. No conversion needed."
          )
      await ctx.info(
          f"Obtaining historical exchange rates for {from_currency} to {to_currency} on {specific_date} from Frankfurter API at {self.frankfurter_api_url}"
      )
      cache_key = hashkey(
          self,
          specific_date=specific_date.isoformat(),
          base_currency=from_currency,
          symbols=tuple([to_currency]),
      )
      cache_hit = cache_key in lru_cache
      date_specific_rates, http_response = self._get_historical_exchange_rates(
          specific_date=specific_date.isoformat(),
          base_currency=from_currency,
          symbols=tuple([to_currency]),
      )
      if cache_hit:
          await ctx.info(
              f"Retrieved historical exchange rates for {specific_date} from least-recently used (LRU) cache."
          )
      await ctx.info(f"Converting {amount} of {from_currency} to {to_currency} on {specific_date}")
      if not date_specific_rates or "rates" not in date_specific_rates:
          raise ValueError(  # pragma: no cover
              f"Could not retrieve exchange rates for {from_currency} to {to_currency} for {specific_date}."
          )
      rate = date_specific_rates["rates"].get(to_currency)
      if rate is None:  # pragma: no cover
          raise ValueError(f"Exchange rate for {from_currency} to {to_currency} not found.")
      converted_amount = amount * float(rate)
      result = CurrencyConversionResponse(
          from_currency=from_currency,
          to_currency=to_currency,
          amount=amount,
          converted_amount=converted_amount,
          exchange_rate=rate,
          rate_date=date_specific_rates["date"],
      )
      return self.get_response_content(response=result, http_response=http_response, cached_response=cache_hit)

• src/frankfurtermcp/model.py:8-16 (schema)

  Pydantic BaseModel defining the response schema for currency conversion, including fields: from_currency, to_currency, amount, converted_amount, exchange_rate, and rate_date.

  class CurrencyConversionResponse(BaseModel):
      """Response model for currency conversion."""
  
      from_currency: ISO4217 = Field(description="The ISO 4217 code of the currency to convert from.")
      to_currency: ISO4217 = Field(description="The ISO 4217 code of the currency to convert to.")
      amount: PositiveFloat = Field(description="The amount (of the source currency) to convert.")
      converted_amount: PositiveFloat = Field(description="The converted amount (of the target currency).")
      exchange_rate: PositiveFloat = Field(description="The exchange rate used for the conversion.")
      rate_date: date = Field(description="The date, in ISO format, of the exchange rate used for the conversion.")

• src/frankfurtermcp/server.py:63-74 (registration)

  Tool registration entry in the FrankfurterMCP.tools class variable, defining the tool name 'convert_currency_specific_date' with tags for currency-rates, currency-conversion, and historical-exchange-rates, plus readOnlyHint and openWorldHint annotations.

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

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

  Internal helper method _get_historical_exchange_rates that fetches historical exchange rates from the Frankfurter API. Uses LRU cache for performance and supports specific_date, start_date, end_date parameters.

  @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}")