cerebrus_depeg

Instructions

Implementation Reference

• src/cerebrus_pulse_mcp/server.py:328-341 (registration)

  Tool 'cerebrus_depeg' is registered in the list_tools() function with its name, description, and empty inputSchema (no required params).

  Tool(
      name="cerebrus_depeg",
      description=(
          "Get USDC collateral health monitor via Chainlink oracle. "
          "Checks USDC/USD deviation from $1.00 peg, reports peg status "
          "(HEALTHY/ELEVATED/WARNING/CRITICAL), risk level, and Arbitrum "
          "sequencer status. Essential before sizing USDC-margined positions. "
          "Cost: $0.01 USDC via x402."
      ),
      inputSchema={
          "type": "object",
          "properties": {},
      },
  ),

• src/cerebrus_pulse_mcp/server.py:423-424 (handler)

  Handler for 'cerebrus_depeg' tool: makes a GET request to the '/depeg' API endpoint with no parameters.

  elif name == "cerebrus_depeg":
      result = _api_get("/depeg")

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

  Helper function _api_get() used to make the HTTP GET request to the Cerebrus Pulse API, which handles 402 (payment_required), 429 (rate_limited), 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:48-49 (helper)

  Helper function _format_response() used to serialize the API response to JSON for the TextContent output.

  def _format_response(data: dict | list) -> str:
      return json.dumps(data, indent=2)