FrankfurterMCP

Instructions

Implementation Reference

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

  The main handler function that implements the convert_currency_latest tool. It accepts amount, from_currency, and to_currency parameters, fetches the latest exchange rates using the helper function, performs the conversion calculation, and returns a CurrencyConversionResponse.

  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-16 (schema)

  The CurrencyConversionResponse Pydantic model that defines the output schema for currency conversion results, including fields for 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:47-54 (registration)

  Tool registration metadata for convert_currency_latest, defining tags (currency-rates, currency-conversion) and annotations (readOnlyHint, openWorldHint) that configure how the tool is exposed via MCP.

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

• src/frankfurtermcp/server.py:105-126 (helper)

  The _get_latest_exchange_rates helper function used by convert_currency_latest to fetch exchange rates from the Frankfurter API. It is cached with TTL caching and supports optional base_currency and symbols filtering.

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

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

  The register_features method in MCPMixin that iterates through the tools list and registers each tool with the FastMCP instance using the mcp.tool() decorator, connecting the tool metadata to the actual handler function.

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