FrankfurterMCP

Instructions

Implementation Reference

• src/frankfurtermcp/server.py:196-242 (handler)

  The async handler function that performs the currency conversion using latest exchange rates. It fetches rates via helper if needed, checks for same currencies, computes converted amount, and returns CurrencyConversionResponse wrapped in response content.
  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/model.py:8-17 (schema)

  Pydantic BaseModel defining the structured output for currency conversions, including from/to currencies, 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:46-53 (registration)

  Tool registration dictionary in the class tools list, specifying fn='convert_currency_latest', tags, and readOnly/openWorld annotations.
  { "fn": "convert_currency_latest", "tags": ["currency-rates", "currency-conversion"], "annotations": { "readOnlyHint": True, "openWorldHint": True, }, },
• src/frankfurtermcp/server.py:96-118 (helper)

  Internal cached helper to fetch latest exchange rates from Frankfurter API /latest endpoint, used by convert_currency_latest.
  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}")