get_candle_snapshot
Retrieve historical candle data for cryptocurrency trading analysis on Hyperliquid DEX. Specify coin symbol and time interval to access market price history for informed decision-making.
Instructions
Get historical candle data for a specific coin
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| coin | Yes | The coin symbol (e.g., BTC, ETH, SOL) | |
| endTime | No | End time in milliseconds (optional) | |
| interval | Yes | Candle interval | |
| startTime | No | Start time in milliseconds (optional) |
Implementation Reference
- The main handler function for the 'get_candle_snapshot' tool. It parses input arguments, calls the Hyperliquid client to fetch candle data, formats the response as text, and returns it in MCP content format.async def handle_get_candle_snapshot(client: HyperliquidClient, args: Dict[str, Any]) -> Dict[str, Any]: """Handle get candle snapshot request.""" coin = args["coin"] interval = args["interval"] start_time = args.get("startTime") end_time = args.get("endTime") result = await client.get_candle_snapshot(coin, interval, start_time, end_time) if not result.success: raise ValueError(f"Failed to get candle data for {coin}: {result.error}") candles = result.data.get("candles", []) if result.data else [] candle_text = "\n".join( f"{candle['t']}: O:{candle['o']} H:{candle['h']} L:{candle['l']} C:{candle['c']} V:{candle['v']}" for candle in candles ) return { "content": [ TextContent( type="text", text=f"Candle data for {coin} ({interval}):\n{candle_text}", ) ] }
- The Tool object definition for 'get_candle_snapshot', including name, description, and detailed input schema with required fields (coin, interval) and optional timestamps.get_candle_snapshot_tool = Tool( name="get_candle_snapshot", description="Get historical candle data for a specific coin", inputSchema={ "type": "object", "properties": { "coin": { "type": "string", "description": "The coin symbol (e.g., BTC, ETH, SOL)", }, "interval": { "type": "string", "description": "Candle interval", "enum": ["1m", "5m", "15m", "1h", "4h", "1d", "1w", "1M"], }, "startTime": { "type": "number", "description": "Start time in milliseconds (optional)", }, "endTime": { "type": "number", "description": "End time in milliseconds (optional)", }, }, "required": ["coin", "interval"], }, )
- Supporting client method in HyperliquidClient that performs the actual API request to Hyperliquid's /info endpoint for candleSnapshot data.async def get_candle_snapshot( self, coin: str, interval: str, start_time: Optional[int] = None, end_time: Optional[int] = None, ) -> ApiResponse[CandleSnapshotResponse]: """Get historical candle data.""" try: req_data = {"coin": coin, "interval": interval} if start_time is not None: req_data["startTime"] = start_time if end_time is not None: req_data["endTime"] = end_time payload = {"type": "candleSnapshot", "req": req_data} response = await self.client.post("/info", json=payload) response.raise_for_status() return ApiResponse(success=True, data=response.json()) except Exception as e: return ApiResponse(success=False, error=str(e))
- hyperliquid_mcp_server/mcp_http_server.py:43-56 (registration)Registration of the tool handler in the HTTP MCP server's TOOL_HANDLERS dictionary for dispatching tool calls.# Map tool names to handlers TOOL_HANDLERS = { "get_all_mids": handle_get_all_mids, "get_l2_book": handle_get_l2_book, "get_candle_snapshot": handle_get_candle_snapshot, "get_open_orders": handle_get_open_orders, "get_user_fills": handle_get_user_fills, "get_user_fills_by_time": handle_get_user_fills_by_time, "get_portfolio": handle_get_portfolio, "place_order": handle_place_order, "place_trigger_order": handle_place_trigger_order, "cancel_order": handle_cancel_order, "cancel_all_orders": handle_cancel_all_orders, }
- hyperliquid_mcp_server/main.py:98-99 (registration)Dispatch logic in the stdio MCP server's call_tool method that routes 'get_candle_snapshot' calls to the handler.elif name == "get_candle_snapshot": result = await handle_get_candle_snapshot(client, args)