execute_hg_dml_sql
Execute INSERT, UPDATE, and DELETE SQL queries to modify data in Hologres databases through the MCP server interface.
Instructions
Execute (INSERT, UPDATE, DELETE) SQL to insert, update, and delete data in Hologres databse.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | The DML SQL query to execute in Hologres database |
Implementation Reference
- src/hologres_mcp_server/server.py:376-389 (registration)Registration of the execute_hg_dml_sql tool in list_tools(), including name, description, and input schema.
Tool( name="execute_hg_dml_sql", description="Execute (INSERT, UPDATE, DELETE) SQL to insert, update, and delete data in Hologres databse.", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The DML SQL query to execute in Hologres database" } }, "required": ["query"] } ), - Input schema definition for execute_hg_dml_sql tool: requires a 'query' string that is the DML SQL.
inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The DML SQL query to execute in Hologres database" } }, "required": ["query"] } - Handler dispatch block in @app.call_tool(): validates that the tool name matches and query is a valid DML statement (INSERT, UPDATE, DELETE). Sets 'query' variable for execution.
elif name == "execute_hg_dml_sql": query = arguments.get("query") if not query: raise ValueError("Query is required") if not any(query.strip().upper().startswith(keyword) for keyword in ["INSERT", "UPDATE", "DELETE"]): raise ValueError("Query must be a DML statement (INSERT, UPDATE, DELETE)") - Shared helper function handle_call_tool that performs the actual database connection, query execution, and result formatting. For DML tools like execute_hg_dml_sql, it executes the query and returns a success message (falls to 'else' clause due to name mismatch).
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)}"