Spaces:
Sleeping
Sleeping
| import json | |
| import requests | |
| from fastapi import FastAPI, HTTPException | |
| from fastapi.middleware.cors import CORSMiddleware | |
| from pydantic import BaseModel | |
| from typing import Any, Dict, Optional | |
| import uvicorn | |
| # Hyperliquid API base URL | |
| HYPERLIQUID_API = "https://api.hyperliquid.xyz/info" | |
| # FastAPI app | |
| app = FastAPI( | |
| title="Hyperliquid MCP Server", | |
| description="MCP server providing real-time trading data from Hyperliquid decentralized exchange", | |
| version="1.0.0" | |
| ) | |
| # Add CORS middleware | |
| app.add_middleware( | |
| CORSMiddleware, | |
| allow_origins=["*"], | |
| allow_credentials=True, | |
| allow_methods=["*"], | |
| allow_headers=["*"], | |
| ) | |
| # Tool definitions | |
| TOOLS = [ | |
| { | |
| "name": "get_all_mids", | |
| "description": "Get all market prices from Hyperliquid", | |
| "inputSchema": {"type": "object", "properties": {}} | |
| }, | |
| { | |
| "name": "get_user_state", | |
| "description": "Get user account state and positions", | |
| "inputSchema": { | |
| "type": "object", | |
| "properties": { | |
| "address": {"type": "string", "description": "User wallet address"} | |
| }, | |
| "required": ["address"] | |
| } | |
| }, | |
| { | |
| "name": "get_recent_trades", | |
| "description": "Get recent trades for a specific coin", | |
| "inputSchema": { | |
| "type": "object", | |
| "properties": { | |
| "coin": {"type": "string", "description": "Trading pair symbol (e.g., 'BTC')"}, | |
| "n": {"type": "number", "description": "Number of trades to retrieve (1-1000)", "default": 100} | |
| }, | |
| "required": ["coin"] | |
| } | |
| }, | |
| { | |
| "name": "get_l2_snapshot", | |
| "description": "Get L2 order book snapshot", | |
| "inputSchema": { | |
| "type": "object", | |
| "properties": { | |
| "coin": {"type": "string", "description": "Trading pair symbol (e.g., 'BTC')"} | |
| }, | |
| "required": ["coin"] | |
| } | |
| }, | |
| { | |
| "name": "get_candles", | |
| "description": "Get historical candlestick data", | |
| "inputSchema": { | |
| "type": "object", | |
| "properties": { | |
| "coin": {"type": "string", "description": "Trading pair symbol (e.g., 'BTC')"}, | |
| "interval": {"type": "string", "description": "Time interval (e.g., '1m', '5m', '1h', '1d')", "default": "1h"}, | |
| "limit": {"type": "number", "description": "Number of candles to retrieve (1-5000)", "default": 500} | |
| }, | |
| "required": ["coin"] | |
| } | |
| }, | |
| { | |
| "name": "get_meta", | |
| "description": "Get market metadata", | |
| "inputSchema": {"type": "object", "properties": {}} | |
| }, | |
| { | |
| "name": "get_funding_rates", | |
| "description": "Get funding rates for perpetual contracts", | |
| "inputSchema": { | |
| "type": "object", | |
| "properties": { | |
| "coin": {"type": "string", "description": "Trading pair symbol (optional)"} | |
| } | |
| } | |
| }, | |
| { | |
| "name": "get_open_interest", | |
| "description": "Get open interest data", | |
| "inputSchema": { | |
| "type": "object", | |
| "properties": { | |
| "coin": {"type": "string", "description": "Trading pair symbol"} | |
| }, | |
| "required": ["coin"] | |
| } | |
| } | |
| ] | |
| # Request models | |
| class ToolCallRequest(BaseModel): | |
| name: str | |
| arguments: Optional[Dict[str, Any]] = {} | |
| # Tool implementations | |
| def get_all_mids(): | |
| """Get all market prices from Hyperliquid""" | |
| try: | |
| response = requests.post(HYPERLIQUID_API, json={"type": "allMids"}) | |
| response.raise_for_status() | |
| return {"success": True, "data": response.json()} | |
| except Exception as e: | |
| return {"success": False, "error": str(e)} | |
| def get_user_state(address: str): | |
| """Get user account state and positions""" | |
| try: | |
| response = requests.post(HYPERLIQUID_API, json={ | |
| "type": "userState", | |
| "user": address | |
| }) | |
| response.raise_for_status() | |
| return {"success": True, "data": response.json()} | |
| except Exception as e: | |
| return {"success": False, "error": str(e)} | |
| def get_recent_trades(coin: str, n: int = 100): | |
| """Get recent trades for a specific coin""" | |
| try: | |
| response = requests.post(HYPERLIQUID_API, json={ | |
| "type": "trades", | |
| "coin": coin, | |
| "n": n | |
| }) | |
| response.raise_for_status() | |
| return {"success": True, "data": response.json()} | |
| except Exception as e: | |
| return {"success": False, "error": str(e)} | |
| def get_l2_snapshot(coin: str): | |
| """Get L2 order book snapshot""" | |
| try: | |
| response = requests.post(HYPERLIQUID_API, json={ | |
| "type": "l2Book", | |
| "coin": coin | |
| }) | |
| response.raise_for_status() | |
| return {"success": True, "data": response.json()} | |
| except Exception as e: | |
| return {"success": False, "error": str(e)} | |
| def get_candles(coin: str, interval: str = "1h", limit: int = 500): | |
| """Get historical candlestick data""" | |
| try: | |
| response = requests.post(HYPERLIQUID_API, json={ | |
| "type": "candles", | |
| "coin": coin, | |
| "interval": interval, | |
| "limit": limit | |
| }) | |
| response.raise_for_status() | |
| return {"success": True, "data": response.json()} | |
| except Exception as e: | |
| return {"success": False, "error": str(e)} | |
| def get_meta(): | |
| """Get market metadata""" | |
| try: | |
| response = requests.post(HYPERLIQUID_API, json={"type": "meta"}) | |
| response.raise_for_status() | |
| return {"success": True, "data": response.json()} | |
| except Exception as e: | |
| return {"success": False, "error": str(e)} | |
| def get_funding_rates(coin: str = None): | |
| """Get funding rates for perpetual contracts""" | |
| try: | |
| payload = {"type": "fundingRate"} | |
| if coin: | |
| payload["coin"] = coin | |
| response = requests.post(HYPERLIQUID_API, json=payload) | |
| response.raise_for_status() | |
| return {"success": True, "data": response.json()} | |
| except Exception as e: | |
| return {"success": False, "error": str(e)} | |
| def get_open_interest(coin: str): | |
| """Get open interest data""" | |
| try: | |
| response = requests.post(HYPERLIQUID_API, json={ | |
| "type": "openInterest", | |
| "coin": coin | |
| }) | |
| response.raise_for_status() | |
| return {"success": True, "data": response.json()} | |
| except Exception as e: | |
| return {"success": False, "error": str(e)} | |
| # API endpoints | |
| async def root(): | |
| """Root endpoint with API information""" | |
| return { | |
| "name": "Hyperliquid MCP Server", | |
| "version": "1.0.0", | |
| "description": "MCP server providing real-time trading data from Hyperliquid decentralized exchange", | |
| "endpoints": { | |
| "health": "/health", | |
| "tools": "/mcp/tools", | |
| "call_tool": "/mcp/call" | |
| } | |
| } | |
| async def health(): | |
| """Health check endpoint""" | |
| return {"status": "healthy", "service": "hyperliquid-mcp-server"} | |
| async def list_tools(): | |
| """List all available MCP tools""" | |
| return {"tools": TOOLS} | |
| async def call_tool(request: ToolCallRequest): | |
| """Execute an MCP tool with provided arguments""" | |
| tool_name = request.name | |
| arguments = request.arguments or {} | |
| # Map tool names to functions | |
| tool_functions = { | |
| "get_all_mids": get_all_mids, | |
| "get_user_state": lambda: get_user_state(arguments.get("address", "")), | |
| "get_recent_trades": lambda: get_recent_trades( | |
| arguments.get("coin", ""), | |
| arguments.get("n", 100) | |
| ), | |
| "get_l2_snapshot": lambda: get_l2_snapshot(arguments.get("coin", "")), | |
| "get_candles": lambda: get_candles( | |
| arguments.get("coin", ""), | |
| arguments.get("interval", "1h"), | |
| arguments.get("limit", 500) | |
| ), | |
| "get_meta": get_meta, | |
| "get_funding_rates": lambda: get_funding_rates(arguments.get("coin")), | |
| "get_open_interest": lambda: get_open_interest(arguments.get("coin", "")) | |
| } | |
| if tool_name not in tool_functions: | |
| raise HTTPException(status_code=404, detail=f"Tool '{tool_name}' not found") | |
| try: | |
| # Execute the tool | |
| if tool_name in ["get_user_state", "get_recent_trades", "get_l2_snapshot", "get_candles", "get_funding_rates", "get_open_interest"]: | |
| result = tool_functions[tool_name]() | |
| else: | |
| result = tool_functions[tool_name]() | |
| if result["success"]: | |
| return { | |
| "content": [ | |
| { | |
| "type": "text", | |
| "text": json.dumps(result["data"], indent=2) | |
| } | |
| ] | |
| } | |
| else: | |
| raise HTTPException(status_code=400, detail=result["error"]) | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| # Update requirements.txt for FastAPI | |
| # fastapi | |
| # uvicorn | |
| # requests | |
| # pydantic | |
| if __name__ == "__main__": | |
| uvicorn.run(app, host="0.0.0.0", port=7860) |