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"]
    }
),

src/hologres_mcp_server/server.py:379-388 (schema)

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"]
}

src/hologres_mcp_server/server.py:566-571 (handler)

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)")

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

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)}"

Hologres MCP Server