execute_sql
Execute SQL queries on Tibero databases to retrieve data, modify records, or inspect schemas through secure database connections.
Instructions
Execute an SQL query on the Tibero server
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | The SQL query to execute |
Implementation Reference
- src/tibero_mcp_server/server.py:212-261 (handler)Handler function for executing SQL queries. Supports SELECT (returns CSV results), DML (INSERT/UPDATE/DELETE with commit and rowcount), and other DDL queries.if name == "execute_sql": query = arguments.get("query") if not query: raise ValueError("Query is required") # Execute the query cursor.execute(query) # Handle different query types query_upper = query.strip().upper() # Data retrieval queries if (query_upper.startswith("SELECT") or query_upper.startswith("SHOW") or query_upper.startswith("DESC")): # Get column names columns = [desc[0] for desc in cursor.description] if cursor.description else [] rows = cursor.fetchall() if not columns: return [TextContent(type="text", text="Query executed successfully, but returned no columns.")] # Format as CSV result = [",".join(map(str, columns))] result.extend([",".join(map(lambda x: str(x) if x is not None else "NULL", row)) for row in rows]) return [TextContent( type="text", text=f"Results ({len(rows)} rows):\n" + "\n".join(result) )] # Non-SELECT queries (DML/DDL) else: # For DML, commit the transaction if (query_upper.startswith("INSERT") or query_upper.startswith("UPDATE") or query_upper.startswith("DELETE")): conn.commit() return [TextContent( type="text", text=f"Query executed successfully. Rows affected: {cursor.rowcount}" )] # For DDL, no commit needed (auto-commit) else: return [TextContent( type="text", text="Query executed successfully." )]
- JSON schema defining the input for the execute_sql tool: an object with a required 'query' string.inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The SQL query to execute" } }, "required": ["query"] }
- src/tibero_mcp_server/server.py:173-186 (registration)Registration of the execute_sql tool in the list_tools() function, including name, description, and input schema.Tool( name="execute_sql", description="Execute an SQL query on the Tibero server", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The SQL query to execute" } }, "required": ["query"] } ),
- Helper function to establish JDBC connection to Tibero database, used by the execute_sql handler.def get_connection(): """Create a JDBC connection to Tibero database.""" config = get_db_config() connection_string = f"jdbc:tibero:thin:@{config['host']}:{config['port']}:{config['sid']}" try: # Make sure the JDBC driver is in the classpath or provide path driver_path = os.getenv("CLASSPATH", "drivers/tibero6-jdbc.jar") conn = jaydebeapi.connect( "com.tmax.tibero.jdbc.TbDriver", connection_string, [config["user"], config["password"]], driver_path ) conn.jconn.setAutoCommit(False) # Set to manual commit mode for safety return conn except Exception as e: logger.error(f"Failed to connect to Tibero: {str(e)}") raise