Hologres MCP Server

Instructions

Implementation Reference

• src/hologres_mcp_server/server.py:584-588 (handler)

  Specific handler logic for get_hg_query_plan tool: extracts the query argument, validates it, and prefixes the SQL with 'EXPLAIN ' before delegating to handle_call_tool.

  elif name == "get_hg_query_plan":
      query = arguments.get("query")
      if not query:
          raise ValueError("Query is required")
      query = f"EXPLAIN {query}"

• src/hologres_mcp_server/server.py:422-435 (registration)

  Registers the 'get_hg_query_plan' tool in the MCP server's list_tools() function, including its name, description, and input schema.

  Tool(
      name="get_hg_query_plan",
      description="Get query plan for a SQL query in Hologres database",
      inputSchema={
          "type": "object",
          "properties": {
              "query": {
                  "type": "string",
                  "description": "The SQL query to analyze in Hologres database"
              }
          },
          "required": ["query"]
      }
  ),

• src/hologres_mcp_server/server.py:425-434 (schema)

  Pydantic input schema definition for the get_hg_query_plan tool, requiring a 'query' string.

  inputSchema={
      "type": "object",
      "properties": {
          "query": {
              "type": "string",
              "description": "The SQL query to analyze in Hologres database"
          }
      },
      "required": ["query"]
  }

• src/hologres_mcp_server/utils.py:48-79 (helper)

  Helper function that executes the prepared SQL query (EXPLAIN <user_query>) on the Hologres database, formats the result as CSV-like text, and handles errors.

  def handle_call_tool(tool_name, query, serverless = False):
      """Handle callTool method."""
      config = get_db_config()
      try:
          with connect_with_retry() as conn:
              with conn.cursor() as cursor:
  
                  # 特殊处理 serverless computing 查询
                  if serverless:
                      cursor.execute("set hg_computing_resource='serverless'")
                  
                  # Execute the query
                  cursor.execute(query)
                  
                  # 特殊处理 ANALYZE 命令
                  if tool_name == "gather_hg_table_statistics":
                      return f"Successfully {query}"
                  
                  # 处理其他有返回结果的查询
                  if cursor.description:  # SELECT query
                      columns = [desc[0] for desc in cursor.description]
                      rows = cursor.fetchall()
                      result = [",".join(map(str, row)) for row in rows]
                      return "\n".join([",".join(columns)] + result)
                  elif tool_name == "execute_dml_sql":  # Non-SELECT query
                      row_count = cursor.rowcount
                      return f"Query executed successfully. {row_count} rows affected."
                  else:
                      return "Query executed successfully"
      except Exception as e:
          return f"Error executing query: {str(e)}"

• src/hologres_mcp_server/utils.py:8-27 (helper)

  Helper function to establish a connection to the Hologres database with retry logic, used by handle_call_tool.

  def connect_with_retry(retries=3):
      attempt = 0
      err_msg = ""
      while attempt <= retries:
          try:
              config = get_db_config()
              conn = psycopg.connect(**config)
              conn.autocommit = True
              with conn.cursor() as cursor:
                  cursor.execute("SELECT 1;")
                  cursor.fetchone()
              return conn
          except psycopg.Error as e:
              err_msg = f"Connection failed: {e}"
              attempt += 1
              if attempt <= retries:
                  print(f"Retrying connection (attempt {attempt + 1} of {retries + 1})...")
                  time.sleep(5)  # 等待 2 秒后再次尝试连接
      raise psycopg.Error(f"Failed to connect to Hologres database after retrying: {err_msg}")