FrankfurterMCP

Instructions

Implementation Reference

• src/frankfurtermcp/server.py:85-98 (handler)

  The main handler function that executes the get_supported_currencies tool logic. It makes an HTTP GET request to the Frankfurter API at /currencies endpoint, decodes the response content, and returns it wrapped in a ToolResult.

  async def get_supported_currencies(self, ctx: Context):
      """Returns a list of three-letter currency codes for the supported currencies."""
      try:
          with self.get_httpx_client() as client:
              await ctx.info(f"Fetching supported currencies from Frankfurter API at {self.frankfurter_api_url}")
              http_response = client.get(f"{self.frankfurter_api_url}/currencies")
              http_response.raise_for_status()
              # Note: The following line could easily be result = http_response.json() but we use content.decode() to
              # demonstrate the TextContent wrapping capability of the get_response_content utility method.
              # Questionable choice? Should we just use # pragma: no cover in the respective branch of get_response_content?
              result = http_response.content.decode()
              return self.get_response_content(response=result, http_response=http_response)
      except httpx.RequestError as e:
          raise ValueError(f"Failed to fetch supported currencies from {self.frankfurter_api_url}. {e}")

• src/frankfurtermcp/server.py:30-38 (registration)

  Tool registration metadata declaring 'get_supported_currencies' as an MCP tool with tags ['currency-rates', 'supported-currencies'] and annotations indicating it's a readOnly, openWorld operation.

  tools = [
      {
          "fn": "get_supported_currencies",
          "tags": ["currency-rates", "supported-currencies"],
          "annotations": {
              "readOnlyHint": True,
              "openWorldHint": True,
          },
      },

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

  The register_features method in MCPMixin that iterates over the tools list and registers each tool with the FastMCP instance using the mcp.tool() decorator. This is where the actual MCP tool registration happens.

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

• src/frankfurtermcp/mixin.py:74-134 (helper)

  The get_response_content helper method used by the handler to convert response data into a ToolResult format with optional metadata. It handles various data types including strings, dicts, lists, and Pydantic models.

  def get_response_content(
      self,
      response: Any,
      http_response: httpx.Response | None = None,
      include_metadata: bool = EnvVar.MCP_SERVER_INCLUDE_METADATA_IN_RESPONSE,
      cached_response: bool = False,
  ) -> ToolResult:
      """Convert response data to a ToolResult format with optional metadata.
  
      Args:
          response (Any): The response data to convert.
          http_response (httpx.Response): The HTTP response object for header extraction.
          include_metadata (bool): Whether to include metadata in the response.
          cached_response (bool): Indicates if the response was served from cache, which will be reflected in metadata.
  
      Returns:
          ToolResult: The ToolResult enclosing the TextContent representation of the response
          along with metadata if requested.
      """
      literal_text = "text"
      text_content: TextContent | None = None
      structured_content: dict[str, Any] | None = None
      if isinstance(response, TextContent):  # pragma: no cover
          text_content = response
          structured_content = {"result": response.text}
      elif isinstance(response, (str, int, float, complex, bool, type(None))):  # pragma: no cover
          text_content = TextContent(type=literal_text, text=str(response))
          structured_content = {"result": response}
      elif isinstance(response, list):  # pragma: no cover
          text_content = TextContent(type=literal_text, text=json.dumps(response))
          structured_content = {"result": response}
      elif isinstance(response, dict):
          structured_content = response
      elif isinstance(response, BaseModel):
          structured_content = response.model_dump()
      else:  # pragma: no cover
          raise TypeError(
              f"Unsupported data type: {type(response).__name__}. "
              "Only str, int, float, complex, bool, dict, list, and Pydantic BaseModel types are supported."
          )
      if text_content is not None:
          tool_result = ToolResult(content=[text_content], structured_content=structured_content)
      elif structured_content is not None:
          tool_result = ToolResult(content=structured_content)
      else:
          assert False, (
              "Unreachable code reached in get_response_content. "
              "Both text_content and structured_content should not have been None."
          )
      if include_metadata:
          tool_result.meta = {
              AppMetadata.PACKAGE_NAME: ResponseMetadata(
                  version=AppMetadata.package_metadata["Version"],
                  api_url=HttpUrl(self.frankfurter_api_url) if http_response else None,
                  api_status_code=http_response.status_code if http_response else None,
                  api_bytes_downloaded=http_response.num_bytes_downloaded if http_response else None,
                  api_elapsed_time=http_response.elapsed.microseconds if http_response else None,
                  cached_response=cached_response,
              ).model_dump(),
          }
      return tool_result

• src/frankfurtermcp/mixin.py:140-155 (helper)

  The get_httpx_client helper method from HTTPHelperMixin that creates and returns an httpx.Client for making HTTP requests to the Frankfurter API, with configurable SSL verification and timeouts.

  def get_httpx_client(self) -> httpx.Client:
      """Obtain an HTTPX client for making requests."""
      verify = EnvVar.HTTPX_VERIFY_SSL
      if verify is False:  # pragma: no cover
          logging.warning("SSL verification is disabled. This is not recommended for production use.")
      ctx = ssl.create_default_context(
          cafile=os.environ.get("SSL_CERT_FILE", certifi.where()),
          capath=os.environ.get("SSL_CERT_DIR"),
      )
      client = httpx.Client(
          verify=verify if (verify is not None and verify is False) else ctx,
          follow_redirects=True,
          trust_env=True,
          timeout=EnvVar.HTTPX_TIMEOUT,
      )
      return client