query_data
Execute SQL queries safely with full schema awareness to interact with databases. Use get_schema() to analyze the database structure before querying.
Instructions
Execute SQL queries safely. Use get_schema() first to understand the database structure.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| sql | Yes |
Implementation Reference
- servers/mcp_server_sqlite.py:128-150 (handler)Handler implementation for query_data tool in SQLite server. Executes arbitrary SQL queries using sqlite3 and formats results.@mcp.tool() def query_data(sql: str) -> str: """Execute SQL queries safely. Use get_schema() first to understand the database structure.""" logger.info(f"Executing SQL query: {sql}") conn = sqlite3.connect(db_path) try: result = conn.execute(sql).fetchall() conn.commit() if not result: return "Query executed successfully but returned no results." # Format results nicely output = f"Query returned {len(result)} row(s):\n\n" for i, row in enumerate(result, 1): output += f"Row {i}: {row}\n" return output except Exception as e: return f"Error: {str(e)}" finally: conn.close()
- servers/mcp_server_mysql.py:185-217 (handler)Handler implementation for query_data tool in MySQL server. Executes SQL queries using mysql.connector, handles SELECT vs mutations differently.@mcp.tool() def query_data(sql: str) -> str: """Execute SQL queries safely. Use get_schema() first to understand the database structure.""" logger.info(f"Executing SQL query: {sql}") try: conn = mysql.connector.connect(**db_config) cursor = conn.cursor() cursor.execute(sql) # Handle different types of queries if sql.strip().upper().startswith(('SELECT', 'SHOW', 'DESCRIBE', 'EXPLAIN')): result = cursor.fetchall() if not result: return "Query executed successfully but returned no results." # Format results nicely output = f"Query returned {len(result)} row(s):\n\n" for i, row in enumerate(result, 1): output += f"Row {i}: {row}\n" else: # For INSERT, UPDATE, DELETE, etc. conn.commit() output = f"Query executed successfully. Affected rows: {cursor.rowcount}" return output except Exception as e: return f"Error: {str(e)}" finally: if 'cursor' in locals(): cursor.close() if 'conn' in locals(): conn.close()
- servers/mcp_server_oracle.py:201-232 (handler)Handler implementation for query_data tool in Oracle server. Executes SQL queries using cx_Oracle, distinguishes SELECT/WITH queries.@mcp.tool() def query_data(sql: str) -> str: """Execute SQL queries safely. Use get_schema() first to understand the database structure.""" logger.info(f"Executing SQL query: {sql}") try: conn = cx_Oracle.connect(**db_config) cursor = conn.cursor() cursor.execute(sql) # Handle different types of queries if sql.strip().upper().startswith(('SELECT', 'WITH')): result = cursor.fetchall() if not result: return "Query executed successfully but returned no results." # Format results nicely output = f"Query returned {len(result)} row(s):\n\n" for i, row in enumerate(result, 1): output += f"Row {i}: {row}\n" else: # For INSERT, UPDATE, DELETE, etc. conn.commit() output = f"Query executed successfully. Affected rows: {cursor.rowcount}" return output except Exception as e: return f"Error: {str(e)}" finally: if 'cursor' in locals(): cursor.close() if 'conn' in locals():