execute_sql
Run SQL queries directly on the OceanBase server to retrieve or manage data efficiently using the specified SQL command.
Instructions
Execute an SQL on the OceanBase server.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| sql | Yes |
Implementation Reference
- The execute_sql tool handler. Executes arbitrary SQL queries on the OceanBase database, returns JSON with results or error.
@app.tool() def execute_sql(sql: str) -> str: """Execute an SQL on the OceanBase server.""" logger.info(f"Calling tool: execute_sql with arguments: {sql}") result = {"sql": sql, "success": False, "rows": 0, "columns": None, "data": None, "error": None} try: with connect(**db_conn_info.model_dump()) as conn: with conn.cursor() as cursor: cursor.execute(sql) if cursor.description: result["columns"] = [column[0] for column in cursor.description] result["data"] = [[str(cell) for cell in row] for row in cursor.fetchall()] else: conn.commit() result["rows"] = cursor.rowcount result["success"] = True except Error as e: result["error"] = f"[Error]: {e}" except Exception as e: result["error"] = f"[Exception]: {e}" json_result = json.dumps(result) if result["error"]: logger.error(f"SQL executed failed, result: {json_result}") return json_result - src/oceanbase_mcp_server/oceanbase_mcp/server.py:162-162 (registration)Registration of the execute_sql tool using FastMCP @app.tool() decorator.
@app.tool()