execute_hg_ddl_sql
Execute CREATE, ALTER, or DROP SQL statements to manage Hologres database objects like tables, views, and procedures.
Instructions
Execute (CREATE, ALTER, DROP) SQL statements to CREATE, ALTER, or DROP tables, views, procedures, GUCs etc. in Hologres databse.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | The DDL SQL query to execute in Hologres database |
Implementation Reference
- src/hologres_mcp_server/server.py:390-403 (registration)Registration of the execute_hg_ddl_sql tool in list_tools(), including description and input schema.Tool( name="execute_hg_ddl_sql", description="Execute (CREATE, ALTER, DROP) SQL statements to CREATE, ALTER, or DROP tables, views, procedures, GUCs etc. in Hologres databse.", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The DDL SQL query to execute in Hologres database" } }, "required": ["query"] } ),
- Input schema definition for the execute_hg_ddl_sql tool: requires a 'query' parameter of type string.inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The DDL SQL query to execute in Hologres database" } }, "required": ["query"] }
- Tool dispatch logic in call_tool(): validates the input query is a DDL statement.elif name == "execute_hg_ddl_sql": query = arguments.get("query") if not query: raise ValueError("Query is required") if not any(query.strip().upper().startswith(keyword) for keyword in ["CREATE", "ALTER", "DROP", "COMMENT ON"]): raise ValueError("Query must be a DDL statement (CREATE, ALTER, DROP, COMMENT ON)")
- src/hologres_mcp_server/utils.py:48-79 (handler)Core handler function that executes the SQL query on Hologres database via psycopg, handles results or success messages for DDL/DML.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)}"
- Helper function for reliable database connection with retries, 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}")