cerebrus_stress

Instructions

Implementation Reference

• src/cerebrus_pulse_mcp/server.py:268-289 (registration)

  Tool definition/registration for 'cerebrus_stress' — declares name, description, and inputSchema (with optional 'limit' param). Registered via the list of Tool objects returned by @server.list_tools().

  Tool(
      name="cerebrus_stress",
      description=(
          "Get market stress index derived from cross-chain arbitrage detection. "
          "Scans 8 chains (Arbitrum, Base, Optimism, Polygon, etc.) for price dislocations. "
          "Returns stress level (LOW/MODERATE/HIGH/EXTREME), score (0-1), "
          "spread statistics, chain routes, and recent scan summaries. "
          "Unique signal — not available from any other provider. Cost: $0.015 USDC via x402."
      ),
      inputSchema={
          "type": "object",
          "properties": {
              "limit": {
                  "type": "integer",
                  "description": "Number of recent scans to analyze (1-50). Default: 10",
                  "default": 10,
                  "minimum": 1,
                  "maximum": 50,
              },
          },
      },
  ),

• src/cerebrus_pulse_mcp/server.py:411-413 (handler)

  Tool handler for 'cerebrus_stress' — extracts the optional 'limit' argument (default 10) and calls _api_get('/arb', params={'limit': limit}). The API path '/arb' corresponds to the cross-chain arbitrage detection endpoint.

  elif name == "cerebrus_stress":
      limit = arguments.get("limit", 10)
      result = _api_get("/arb", params={"limit": limit})

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

  The _api_get helper function used by the tool handler to make GET requests to the Cerebrus Pulse API, handling 402 payment, 429 rate limit, and standard 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:277-288 (schema)

  Input schema for 'cerebrus_stress' — defines a single optional integer parameter 'limit' (1-50, default 10) controlling the number of recent scans to analyze.

  inputSchema={
      "type": "object",
      "properties": {
          "limit": {
              "type": "integer",
              "description": "Number of recent scans to analyze (1-50). Default: 10",
              "default": 10,
              "minimum": 1,
              "maximum": 50,
          },
      },
  },

• src/cerebrus_pulse_mcp/server.py:478-479 (registration)

  CLI registration entry for the stress tool — maps CLI name 'stress' to API path '/arb' with optional limit parameter (default 10).

  "stress":        ("/arb",              [("limit", False, int, 10)]),
  "cex-dex":       ("/cex-dex/{coin}",    [("coin", True, str, None)]),