execute_query
Run SQL queries and retrieve formatted results with each row on a new line, columns separated by tabs, for easy inspection.
Instructions
Execute a SQL query and return the results as a formatted string.
Args: query (str): The SQL query to execute
Returns: str: A formatted string representation of the query results Each row is on a new line, with columns separated by tabs
Raises: Error: If the query execution fails
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
Implementation Reference
- tools/sql_tools.py:182-211 (handler)The execute_query method in the SQLTools class. It executes a SQL query, returns results as a tab-separated formatted string with column headers and rows, or returns an affected-row count for non-SELECT queries.
def execute_query(self, query: str) -> str: """Execute a SQL query and return the results as a formatted string. Args: query (str): The SQL query to execute Returns: str: A formatted string representation of the query results Each row is on a new line, with columns separated by tabs Raises: Error: If the query execution fails """ with self.get_connection() as conn: cursor = conn.cursor(dictionary=True, buffered=True) cursor.execute(query) conn.commit() if cursor.with_rows: results = cursor.fetchall() if not results: return "No results found." columns = list(results[0].keys()) output = "\t".join(columns) + "\n" for row in results: output += "\t".join(str(row[col]) for col in columns) + "\n" return output return f"Query executed successfully. Rows affected: {cursor.rowcount}" - server.py:20-20 (registration)The tool is registered with the MCP server via mcp.tool()(sql_tools.execute_query) on line 20.
mcp.tool()(sql_tools.execute_query) - tools/sql_tools.py:195-196 (helper)Uses get_connection() (a context manager defined at line 19) to obtain a database connection with dictionary cursor and buffered mode.
with self.get_connection() as conn: cursor = conn.cursor(dictionary=True, buffered=True) - test_sql_tools.py:111-115 (helper)Test case for execute_query with SELECT 1 to verify the results contain column names and values.
def test_execute_query_select_one(sql_tools: SQLTools): result = sql_tools.execute_query("SELECT 1 AS test_column") assert isinstance(result, str) assert "test_column" in result assert "1" in result - test_sql_tools.py:118-121 (helper)Test case for execute_query with SELECT VERSION() to verify MySQL version retrieval.
def test_execute_query_version(sql_tools: SQLTools): result = sql_tools.execute_query("SELECT VERSION() AS mysql_version") assert isinstance(result, str) assert "mysql_version" in result