Financial Datasets MCP Server

Instructions

Args:
    ticker: Ticker symbol of the crypto currency (e.g. BTC-USD). The list of available crypto tickers can be retrieved via the get_available_crypto_tickers tool.

Implementation Reference

• server.py:311-334 (handler)

  The main handler function for the 'get_current_crypto_price' tool. It fetches the current price snapshot for a given crypto ticker from the Financial Datasets API using the shared make_request helper, handles errors, and returns a JSON string of the snapshot.

  @mcp.tool()
  async def get_current_crypto_price(ticker: str) -> str:
      """Get the current / latest price of a crypto currency.
  
      Args:
          ticker: Ticker symbol of the crypto currency (e.g. BTC-USD). The list of available crypto tickers can be retrieved via the get_available_crypto_tickers tool.
      """
      # Fetch data from the API
      url = f"{FINANCIAL_DATASETS_API_BASE}/crypto/prices/snapshot/?ticker={ticker}"
      data = await make_request(url)
  
      # Check if data is found
      if not data:
          return "Unable to fetch current price or no current price found."
  
      # Extract the current price
      snapshot = data.get("snapshot", {})
  
      # Check if current price is found
      if not snapshot:
          return "Unable to fetch current price or no current price found."
  
      # Stringify the current price
      return json.dumps(snapshot, indent=2)

• server.py:311-311 (registration)

  The @mcp.tool() decorator registers the get_current_crypto_price function as an MCP tool.

  @mcp.tool()

• server.py:25-41 (helper)

  Shared helper function used by get_current_crypto_price to make authenticated API requests to the Financial Datasets API.

  async def make_request(url: str) -> dict[str, any] | None:
      """Make a request to the Financial Datasets API with proper error handling."""
      # Load environment variables from .env file
      load_dotenv()
      
      headers = {}
      if api_key := os.environ.get("FINANCIAL_DATASETS_API_KEY"):
          headers["X-API-KEY"] = api_key
  
      async with httpx.AsyncClient() as client:
          try:
              response = await client.get(url, headers=headers, timeout=30.0)
              response.raise_for_status()
              return response.json()
          except Exception as e:
              return {"Error": str(e)}