aster-info-mcp

Instructions

Implementation Reference

• main.py:172-245 (handler)

  The handler function that implements the logic for the 'get_mark_price_kline' tool. It fetches Mark Price Kline/Candlestick data from the Aster Finance API endpoint '/fapi/v1/markPriceKlines', processes the JSON response using pandas to create a formatted Markdown table of open_time, open, high, low, and close prices, and handles errors appropriately. The @mcp.tool() decorator registers this function as an MCP tool.
  @mcp.tool() async def get_mark_price_kline( symbol: str, interval: str, startTime: Optional[int] = None, endTime: Optional[int] = None, limit: Optional[int] = None ) -> str: """ Fetch Mark Price Kline/Candlestick data from Aster Finance API and return as Markdown table text. Parameters: symbol (str): Trading pair symbol (e.g., 'BTCUSDT', 'ETHUSDT'). Case-insensitive. interval (str): Kline interval (e.g., '1m' for 1 minute, '1h' for 1 hour, '1d' for 1 day). startTime (Optional[int]): Start time in milliseconds since Unix epoch. If None, defaults to API behavior. endTime (Optional[int]): End time in milliseconds since Unix epoch. If None, defaults to API behavior. limit (Optional[int]): Number of Klines to return (1 to 1500). If None, defaults to 500. Returns: str: Markdown table containing open_time, open, high, low, and close. Raises: Exception: If the API request fails or data processing encounters an error. """ endpoint = "/fapi/v1/markPriceKlines" # Construct query parameters params = { "symbol": symbol.upper(), # Ensure symbol is uppercase (e.g., BTCUSDT) "interval": interval, # e.g., 1m, 1h, 1d } if startTime is not None: params["startTime"] = startTime if endTime is not None: params["endTime"] = endTime if limit is not None: params["limit"] = limit async with httpx.AsyncClient() as client: try: # Make GET request to the API response = await client.get(f"{BASE_URL}{endpoint}", params=params) response.raise_for_status() # Raise exception for 4xx/5xx errors # Parse JSON response kline_data: List[List[Any]] = response.json() # Create pandas DataFrame df = pd.DataFrame(kline_data, columns=[ "open_time", "open", "high", "low", "close", "volume", "close_time", "ignore1", "ignore2", "ignore3", "ignore4", "ignore5" ]) # Convert timestamps to readable format df["open_time"] = pd.to_datetime(df["open_time"], unit="ms") # Select relevant columns and format numbers df = df[["open_time", "open", "high", "low", "close"]] df["open"] = df["open"].astype(float).round(8) df["high"] = df["high"].astype(float).round(8) df["low"] = df["low"].astype(float).round(8) df["close"] = df["close"].astype(float).round(8) # Convert DataFrame to Markdown table markdown_table = df.to_markdown(index=False) return markdown_table except httpx.HTTPStatusError as e: # Handle HTTP errors (e.g., 400, 429) raise Exception(f"API request failed: {e.response.status_code} - {e.response.text}") except Exception as e: # Handle other errors (e.g., network issues, pandas errors) raise Exception(f"Error processing Mark Price Kline data: {str(e)}")
• main.py:172-172 (registration)

  The @mcp.tool() decorator registers the get_mark_price_kline function as an MCP tool in the FastMCP server.
  @mcp.tool()
• main.py:173-179 (schema)

  The function signature defines the input schema with type hints for parameters: symbol (str), interval (str), optional startTime/endTime/limit (int), returning str (Markdown table). This serves as the tool's input/output schema for FastMCP.
  async def get_mark_price_kline( symbol: str, interval: str, startTime: Optional[int] = None, endTime: Optional[int] = None, limit: Optional[int] = None ) -> str: