FrankfurterMCP

Instructions

Implementation Reference

• src/frankfurtermcp/server.py:204-251 (handler)

  The main handler function that performs the currency conversion using the latest exchange rates from Frankfurter API, including input validation via Pydantic annotations, caching, and response formatting.
  async def convert_currency_latest( 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.")], ): """Converts an amount from one currency to another using the latest exchange rates.""" 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 latest exchange rates for {from_currency} to {to_currency} from Frankfurter API at {self.frankfurter_api_url}" ) cache_key = hashkey( self, base_currency=from_currency, symbols=tuple([to_currency]), ) cache_hit = cache_key in ttl_cache latest_rates, http_response = self._get_latest_exchange_rates( base_currency=from_currency, symbols=tuple([to_currency]), ) if cache_hit: await ctx.info("Latest exchange rates fetched from TTL cache.") await ctx.info(f"Converting {amount} of {from_currency} to {to_currency}") if not latest_rates or "rates" not in latest_rates: # pragma: no cover raise ValueError(f"Could not retrieve exchange rates for {from_currency} to {to_currency}.") rate = latest_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=latest_rates["date"], ) return self.get_response_content(response=result, http_response=http_response, cached_response=cache_hit)
• src/frankfurtermcp/server.py:46-53 (registration)

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

  Pydantic BaseModel defining the output schema for currency conversion responses, used by the handler.
  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:104-126 (helper)

  Cached helper function to fetch latest exchange rates from Frankfurter API, called by the convert_currency_latest handler.
  def _get_latest_exchange_rates( self, base_currency: str | None = None, symbols: tuple[str, ...] | None = None, ): """Internal function to get the latest 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) with self.get_httpx_client() as client: http_response = client.get( f"{self.frankfurter_api_url}/latest", 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 latest exchange rates from {self.frankfurter_api_url}. {e}")