Hologres MCP Server

Instructions

Implementation Reference

• src/hologres_mcp_server/server.py:362-375 (registration)

  Registration of the execute_hg_select_sql_with_serverless tool in the list_tools function, including name, description, and input schema.
  Tool( name="execute_hg_select_sql_with_serverless", description="Use Serverless Computing resources to execute SELECT SQL to query data in Hologres database. When the error like \"Total memory used by all existing queries exceeded memory limitation\" occurs during execute_hg_select_sql execution, you can re-execute the SQL with the tool execute_hg_select_sql_with_serverless.", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The (SELECT) SQL query to execute with serverless computing in Hologres database" } }, "required": ["query"] } ),
• src/hologres_mcp_server/server.py:558-565 (handler)

  Handler dispatch logic for the tool in the call_tool function. Validates the query is a SELECT statement and sets the serverless flag to true.
  elif name == "execute_hg_select_sql_with_serverless": query = arguments.get("query") if not query: raise ValueError("Query is required") if not query.strip().upper().startswith("SELECT"): raise ValueError("Query must be a SELECT statement") # 修改 serverless computing 设置方式 serverless = True
• src/hologres_mcp_server/utils.py:48-79 (handler)

  Core handler function that executes the SQL query. For serverless mode (used by this tool), it sets 'hg_computing_resource=\'serverless\'' before executing the query and formats SELECT results as CSV.
  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/server.py:365-374 (schema)

  Input schema definition for the tool, specifying the required 'query' parameter.
  inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The (SELECT) SQL query to execute with serverless computing in Hologres database" } }, "required": ["query"] }