cerebrus_cex_dex

Instructions

Implementation Reference

• src/cerebrus_pulse_mcp/server.py:290-308 (registration)

  Tool registration: cerebrus_cex_dex is registered as a Tool object with name='cerebrus_cex_dex', description about CEX-DEX price divergence, and inputSchema requiring a 'coin' string parameter.

  Tool(
      name="cerebrus_cex_dex",
      description=(
          "Get CEX-DEX price divergence for a token. Compares Coinbase (CEX) vs "
          "Chainlink/Uniswap (DEX) prices. Returns spread in bps, direction "
          "(cex_premium or dex_premium), and interpretation. "
          "Refreshes every 5 minutes. Cost: $0.02 USDC via x402."
      ),
      inputSchema={
          "type": "object",
          "properties": {
              "coin": {
                  "type": "string",
                  "description": "Coin ticker (e.g., ETH, BTC, LINK). Case-insensitive.",
              },
          },
          "required": ["coin"],
      },
  ),

• src/cerebrus_pulse_mcp/server.py:415-417 (handler)

  Tool handler: when name == 'cerebrus_cex_dex', it validates the 'coin' argument via _validate_coin and calls _api_get(f'/cex-dex/{coin}') to fetch the CEX-DEX price divergence data from the backend API.

  elif name == "cerebrus_cex_dex":
      coin = _validate_coin(arguments["coin"])
      result = _api_get(f"/cex-dex/{coin}")

• src/cerebrus_pulse_mcp/server.py:60-83 (helper)

  Helper function: _api_get makes the actual HTTP GET request to the Cerebrus Pulse API, handling 402 (payment required), 429 (rate limiting), and other HTTP errors.

  def _api_get(path: str, params: dict | None = None) -> dict[str, Any]:
      """Make a GET request to the Cerebrus Pulse API."""
      with _make_client() as client:
          resp = client.get(path, params=params)
  
          if resp.status_code == 402:
              # Return payment details so the agent/user knows cost
              return {
                  "status": "payment_required",
                  "message": "This endpoint requires x402 USDC payment on Base or Solana.",
                  "url": f"{BASE_URL}{path}",
                  "payment_details": resp.headers.get("X-Payment", "See x402 SDK docs"),
                  "help": "Install the x402 SDK and set CEREBRUS_WALLET_KEY (Base) or CEREBRUS_WALLET_KEY_SOLANA (Solana) to enable auto-payment. See https://cerebruspulse.xyz/guides/x402-payments",
              }
  
          if resp.status_code == 429:
              return {
                  "status": "rate_limited",
                  "message": "Rate limit exceeded. Back off and retry.",
                  "detail": resp.json() if resp.headers.get("content-type", "").startswith("application/json") else resp.text,
              }
  
          resp.raise_for_status()
          return resp.json()

• src/cerebrus_pulse_mcp/server.py:40-45 (helper)

  Helper function: _make_client creates the httpx.Client configured with the base URL and timeout.

  def _make_client() -> httpx.Client:
      return httpx.Client(
          base_url=BASE_URL,
          timeout=REQUEST_TIMEOUT,
          headers={"User-Agent": f"cerebrus-pulse-mcp/{_VERSION}"},
      )

• src/cerebrus_pulse_mcp/server.py:52-57 (helper)

  Helper function: _validate_coin validates and normalizes the coin ticker input (uppercased, regex-checked).

  def _validate_coin(coin: str) -> str:
      """Validate and normalize a coin ticker. Raises ValueError on bad input."""
      coin = coin.strip().upper()
      if not _COIN_RE.match(coin):
          raise ValueError(f"Invalid coin ticker: {coin!r}")
      return coin