FrankfurterMCP

Instructions

Implementation Reference

• src/frankfurtermcp/server.py:315-375 (handler)

  Handler function that executes the currency conversion logic for a specific historical date using the Frankfurter API.
  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/server.py:62-73 (registration)

  Registration of the convert_currency_specific_date tool in the tools list of the MCP class.
  { "fn": "convert_currency_specific_date", "tags": [ "currency-rates", "currency-conversion", "historical-exchange-rates", ], "annotations": { "readOnlyHint": True, "openWorldHint": True, }, },
• src/frankfurtermcp/model.py:8-17 (schema)

  Pydantic model defining the output schema for the currency conversion response.
  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:119-157 (helper)

  Cached internal helper function to retrieve historical exchange rates from the Frankfurter API, used by the handler.
  @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}")