execute_sql
Execute SQL queries on a MySQL database to retrieve, update, or manage data through secure database interactions.
Instructions
Execute an SQL query on the MySQL server
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | The SQL query to execute |
Implementation Reference
- src/mysql_mcp_server/server.py:110-149 (handler)The main handler for the execute_sql tool. It validates the tool name and query argument, connects to the MySQL database, executes the SQL query, handles results for SELECT-like queries (including special SHOW TABLES), commits non-query changes, and returns formatted text results or error messages.async def call_tool(name: str, arguments: dict) -> list[TextContent]: """Execute SQL commands.""" config = get_db_config() logger.info(f"Calling tool: {name} with arguments: {arguments}") if name != "execute_sql": raise ValueError(f"Unknown tool: {name}") query = arguments.get("query") if not query: raise ValueError("Query is required") try: with connect(**config) as conn: with conn.cursor() as cursor: cursor.execute(query) # Special handling for SHOW TABLES if query.strip().upper().startswith("SHOW TABLES"): tables = cursor.fetchall() result = ["Tables_in_" + config["database"]] # Header result.extend([table[0] for table in tables]) return [TextContent(type="text", text="\n".join(result))] # Handle all other queries that return result sets (SELECT, SHOW, DESCRIBE etc.) elif cursor.description: 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))] # Non-SELECT queries else: conn.commit() return [TextContent(type="text", text=f"Query executed successfully. Rows affected: {cursor.rowcount}")] except Error as e: logger.error(f"Error executing SQL '{query}': {e}") return [TextContent(type="text", text=f"Error executing query: {str(e)}")]
- Pydantic-like input schema definition for the execute_sql tool, requiring a 'query' string property.inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The SQL query to execute" } }, "required": ["query"] }
- src/mysql_mcp_server/server.py:92-107 (registration)The execute_sql tool is registered in the list_tools() decorator function by returning a Tool object with name, description, and inputSchema.return [ Tool( name="execute_sql", description="Execute an SQL query on the MySQL server", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The SQL query to execute" } }, "required": ["query"] } ) ]
- src/mysql_mcp_server/server.py:16-31 (helper)Helper function to retrieve MySQL database connection configuration from environment variables, used by the handler.def get_db_config(): """Get database configuration from environment variables.""" config = { "host": os.getenv("MYSQL_HOST", "localhost"), "port": int(os.getenv("MYSQL_PORT", "3306")), "user": os.getenv("MYSQL_USER"), "password": os.getenv("MYSQL_PASSWORD"), "database": os.getenv("MYSQL_DATABASE") } if not all([config["user"], config["password"], config["database"]]): logger.error("Missing required database configuration. Please check environment variables:") logger.error("MYSQL_USER, MYSQL_PASSWORD, and MYSQL_DATABASE are required") raise ValueError("Missing required database configuration") return config