search_crypto_coins

Instructions

Implementation Reference

• app/tools/data.py:431-470 (handler)

  The @mcp.tool decorated function 'search_crypto_coins' that serves as the MCP tool handler. It accepts a query string, validates it, calls search_crypto(), and returns a dict with query, coins list, count, and fetched_at timestamp.

  @mcp.tool
  def search_crypto_coins(query: str) -> dict[str, Any]:
      """Search for cryptocurrencies on CoinGecko.
  
      Find coins by name, symbol, or keyword. Useful for discovering
      coin IDs to use with create_portfolio.
  
      Args:
          query: Search query (e.g., 'bitcoin', 'defi', 'layer 2').
  
      Returns:
          Dictionary containing:
          - coins: List of matching coins (id, name, symbol, market_cap_rank)
          - count: Number of results
          - fetched_at: ISO timestamp
  
      Example:
          ```
          # Search for DeFi coins
          result = search_crypto_coins(query="defi")
          for coin in result['coins']:
              print(f"{coin['name']} ({coin['symbol']}) - Rank: {coin['market_cap_rank']}")
          ```
      """
      if not query or len(query.strip()) == 0:
          return {"error": "Query cannot be empty"}
  
      try:
          coins = search_crypto(query.strip())
          return {
              "query": query,
              "coins": coins,
              "count": len(coins),
              "fetched_at": datetime.now().isoformat(),
          }
      except ValueError as error:
          return {
              "error": str(error),
              "suggestion": "CoinGecko API may be rate-limited. Try again later.",
          }

• app/data_sources.py:439-458 (helper)

  Helper function 'search_crypto' that performs the actual CoinGecko API call via _coingecko_request('search', ...) and returns a list of up to 20 matching coins with id, name, symbol, and market_cap_rank.

  def search_crypto(query: str) -> list[dict[str, Any]]:
      """Search for cryptocurrencies on CoinGecko.
  
      Args:
          query: Search query (e.g., 'bitcoin', 'defi').
  
      Returns:
          List of matching coins.
      """
      data = _coingecko_request("search", params={"query": query})
  
      return [
          {
              "id": coin["id"],
              "name": coin["name"],
              "symbol": coin["symbol"],
              "market_cap_rank": coin.get("market_cap_rank"),
          }
          for coin in data.get("coins", [])[:20]  # Limit to top 20
      ]

• app/data_sources.py:90-118 (helper)

  Core API helper '_coingecko_request' that calls CoinGecko API with rate limiting and error handling, used by search_crypto and all other CoinGecko functions.

  def _coingecko_request(
      endpoint: str,
      params: dict[str, Any] | None = None,
  ) -> dict[str, Any]:
      """Make a rate-limited request to CoinGecko API.
  
      Args:
          endpoint: API endpoint (without base URL).
          params: Query parameters.
  
      Returns:
          JSON response as dictionary.
  
      Raises:
          ValueError: If rate limited or request fails.
      """
      url = f"{COINGECKO_API_URL}/{endpoint}"
      try:
          response = requests.get(url, params=params, timeout=30)
          if response.status_code == 429:
              logger.warning("CoinGecko rate limit hit")
              raise ValueError(
                  "CoinGecko rate limit exceeded. Try again in a minute or use fewer symbols."
              )
          response.raise_for_status()
          return response.json()
      except requests.exceptions.RequestException as error:
          logger.error(f"CoinGecko request failed: {error}")
          raise ValueError(f"Failed to fetch crypto data: {error}") from error

• app/tools/data.py:34-36 (registration)

  The 'register_data_tools' function that registers all data tools (including search_crypto_coins via @mcp.tool decorator) with the FastMCP server. The @mcp.tool decorator on the function at line 431 handles the actual registration.

  def register_data_tools(
      mcp: FastMCP, store: PortfolioStore, cache: RefCache | None = None
  ) -> None:

• app/tools/data.py:432-454 (schema)

  The tool's docstring acts as the schema/interface definition, describing the single 'query' input parameter (str), and the return dict structure containing 'coins', 'count', and 'fetched_at'.

  def search_crypto_coins(query: str) -> dict[str, Any]:
      """Search for cryptocurrencies on CoinGecko.
  
      Find coins by name, symbol, or keyword. Useful for discovering
      coin IDs to use with create_portfolio.
  
      Args:
          query: Search query (e.g., 'bitcoin', 'defi', 'layer 2').
  
      Returns:
          Dictionary containing:
          - coins: List of matching coins (id, name, symbol, market_cap_rank)
          - count: Number of results
          - fetched_at: ISO timestamp
  
      Example:
          ```
          # Search for DeFi coins
          result = search_crypto_coins(query="defi")
          for coin in result['coins']:
              print(f"{coin['name']} ({coin['symbol']}) - Rank: {coin['market_cap_rank']}")
          ```
      """