execute_hg_select_sql
Query data from Hologres databases using SELECT SQL statements to retrieve specific information for analysis and decision-making.
Instructions
Execute SELECT SQL to query data from Hologres database.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | The (SELECT) SQL query to execute in Hologres database. |
Implementation Reference
- src/hologres_mcp_server/utils.py:48-79 (handler)Core handler function that executes the SQL query for the execute_hg_select_sql tool (and others). Connects to Hologres database, executes the query, formats SELECT results as CSV-like text with columns and rows.
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:347-360 (registration)Registration of the execute_hg_select_sql tool in the MCP server's list_tools() function, including name, description, and input schema.
Tool( name="execute_hg_select_sql", description="Execute SELECT SQL to query data from Hologres database.", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The (SELECT) SQL query to execute in Hologres database." } }, "required": ["query"] } ), - Specific dispatching and input validation logic for the execute_hg_select_sql tool within the call_tool handler, ensuring it's a SELECT query before passing to execution utility.
if name == "execute_hg_select_sql": query = arguments.get("query") if not query: raise ValueError("Query is required") if not re.match(r"^\s*WITH\s+.*?SELECT\b", query, re.IGNORECASE) and not re.match(r"^\s*SELECT\b", query, re.IGNORECASE): raise ValueError("Query must be a SELECT statement or start with WITH followed by a SELECT statement") - Helper function used by the handler to establish a reliable database connection to Hologres with retries.
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}")