cerebrus_pulse

Instructions

Implementation Reference

• src/cerebrus_pulse_mcp/server.py:113-137 (registration)

  Tool 'cerebrus_pulse' is registered as an MCP tool with name, description and inputSchema (requires 'coin', optional 'timeframes').

  Tool(
      name="cerebrus_pulse",
      description=(
          "Get multi-timeframe technical analysis for a Hyperliquid perpetual. "
          "Supports 6 timeframes: 5m, 15m, 1h, 4h, 1d, 1w (daily/weekly aggregated from 1h). "
          "Returns RSI, EMAs (20/50/200), ATR, Bollinger Bands, VWAP, Z-score, "
          "trend direction, cross-timeframe confluence with alignment scoring, "
          "derivatives data (funding, OI, spread), and market regime. Cost: $0.02 USDC via x402."
      ),
      inputSchema={
          "type": "object",
          "properties": {
              "coin": {
                  "type": "string",
                  "description": "Coin ticker (e.g., BTC, ETH, SOL). Case-insensitive.",
              },
              "timeframes": {
                  "type": "string",
                  "description": "Comma-separated timeframes: 5m, 15m, 1h, 4h, 1d, 1w. Default: 1h,4h",
                  "default": "1h,4h",
              },
          },
          "required": ["coin"],
      },
  ),

• src/cerebrus_pulse_mcp/server.py:378-381 (handler)

  Handler for 'cerebrus_pulse': validates coin via _validate_coin, gets optional timeframes (default '1h,4h'), then calls _api_get('/pulse/{coin}', params={'timeframes': ...}).

  elif name == "cerebrus_pulse":
      coin = _validate_coin(arguments["coin"])
      timeframes = arguments.get("timeframes", "1h,4h")
      result = _api_get(f"/pulse/{coin}", params={"timeframes": timeframes})

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

  Helper _validate_coin: validates/normalizes coin ticker (uppercases, checks regex).

  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

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

  Helper _api_get: makes GET request to cerebro Pulse API, handles 402/429 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:467-468 (registration)

  CLI mapping for 'pulse' tool (short name) registered in _CLI_TOOLS dict.

  "pulse":         ("/pulse/{coin}",      [("coin", True, str, None),
                                           ("timeframes", False, str, "1h,4h")]),