execute_sql
Run SQL queries directly on the Adb MySQL Cluster using this tool, designed for streamlined database interaction and data retrieval.
Instructions
Execute a SQL query in the Adb MySQL Cluster
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | The SQL query to execute |
Implementation Reference
- The main handler function for MCP tools, including 'execute_sql'. For 'execute_sql', it retrieves the SQL query from arguments, connects to the Adb MySQL database using pymysql, executes the query, formats the results as CSV-like text, and returns it as TextContent. Error handling and connection cleanup are included.@app.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: """Execute SQL commands.""" config = get_db_config() if name == "execute_sql": query = arguments.get("query") if not query: raise ValueError("Query is required") elif name == "get_query_plan": query = arguments.get("query") if not query: raise ValueError("Query is required") query = f"EXPLAIN {query}" elif name == "get_execution_plan": query = arguments.get("query") if not query: raise ValueError("Query is required") query = f"EXPLAIN ANALYZE {query}" else: raise ValueError(f"Unknown tool: {name}") conn = pymysql.connect(**config) conn.autocommit(True) cursor = conn.cursor() try: # Execute the query cursor.execute(query) columns = [desc[0] for desc in cursor.description] rows = cursor.fetchall() result = [",".join(map(str, row)) for row in rows] return [TextContent(type="text", text="\n".join([",".join(columns)] + result))] except Exception as e: return [TextContent(type="text", text=f"Error executing query: {str(e)}")] finally: if cursor: cursor.close() if conn.open: conn.close()
- Input schema definition for the 'execute_sql' tool, specifying an object with a required 'query' string property."type": "object", "properties": { "query": { "type": "string", "description": "The SQL query to execute" } }, "required": ["query"] },
- src/adb_mysql_mcp_server/server.py:123-136 (registration)Registration of the 'execute_sql' tool in the list_tools() function, including name, description, and input schema.Tool( name="execute_sql", description="Execute a SQL query in the Adb MySQL Cluster", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The SQL query to execute" } }, "required": ["query"] }, ),
- Helper function to load database connection configuration from environment variables, used by the execute_sql handler.def get_db_config(): config = { "host": os.getenv("ADB_MYSQL_HOST", "localhost"), "port": int(os.getenv("ADB_MYSQL_PORT", 3306)), "user": os.getenv("ADB_MYSQL_USER"), "password": os.getenv("ADB_MYSQL_PASSWORD"), "database": os.getenv("ADB_MYSQL_DATABASE"), } if not all([config["user"], config["password"], config["database"]]): raise ValueError("Missing required database configuration") return config