Oracle MCP Server

Instructions

Implementation Reference

• src/oracle_mcp_server/server.py:371-464 (handler)

  Core handler that executes the SQL query: performs safety checks (blocks DML/DDL), adds automatic ROWNUM row limits for SELECTs, executes via oracledb.Connection, handles Oracle-specific types (LOBs, dates), and returns structured JSON results.
  async def execute_query( self, sql: str, params: Optional[List] = None ) -> Dict[str, Any]: """Execute a SQL query with safety controls""" # Basic SQL injection prevention sql_upper = sql.upper().strip() # Check for potentially dangerous operations dangerous_keywords = [ "DROP", "DELETE", "TRUNCATE", "ALTER", "CREATE", "INSERT", "UPDATE", ] # Allow SELECT, DESCRIBE, EXPLAIN PLAN if not any( sql_upper.startswith(keyword) for keyword in ["SELECT", "WITH", "DESCRIBE", "DESC", "EXPLAIN"] ): if any(keyword in sql_upper for keyword in dangerous_keywords): raise ValueError( "Only SELECT, DESCRIBE, and EXPLAIN PLAN statements are allowed" ) conn = await self.connection_manager.get_connection() try: cursor = conn.cursor() # Set row limit if ( "SELECT" in sql_upper and "ROWNUM" not in sql_upper and "LIMIT" not in sql_upper ): # Add ROWNUM limitation for SELECT queries if "ORDER BY" in sql_upper: # More complex query, wrap it sql = f"SELECT * FROM ({sql}) WHERE ROWNUM <= {QUERY_LIMIT_SIZE}" else: # Simple query, add WHERE clause if "WHERE" in sql_upper: sql += f" AND ROWNUM <= {QUERY_LIMIT_SIZE}" else: sql += f" WHERE ROWNUM <= {QUERY_LIMIT_SIZE}" start_time = datetime.now() if params: cursor.execute(sql, params) else: cursor.execute(sql) execution_time = (datetime.now() - start_time).total_seconds() # Fetch results if cursor.description: columns = [desc[0] for desc in cursor.description] rows = cursor.fetchall() # Convert Oracle types to JSON-serializable types serializable_rows = [] for row in rows: serializable_row = [] for value in row: if hasattr(value, "read"): # LOB object serializable_row.append(str(value.read())) elif isinstance(value, datetime): serializable_row.append(value.isoformat()) else: serializable_row.append(value) serializable_rows.append(serializable_row) return { "columns": columns, "rows": serializable_rows, "row_count": len(rows), "execution_time_seconds": execution_time, "query": sql, } else: return { "message": "Query executed successfully", "execution_time_seconds": execution_time, "query": sql, } finally: conn.close()
• src/oracle_mcp_server/server.py:628-647 (registration)

  Registers the execute_query tool in the MCP server's list_tools() handler, defining its metadata and input schema.
  Tool( name="execute_query", description="Execute a SQL query against the Oracle database. Only SELECT, DESCRIBE, and EXPLAIN PLAN statements are allowed for safety.", inputSchema={ "type": "object", "properties": { "sql": { "type": "string", "description": "SQL query to execute (SELECT, DESCRIBE, or EXPLAIN PLAN only)", }, "params": { "type": "array", "items": {"type": "string"}, "description": "Optional parameters for parameterized queries", "default": [], }, }, "required": ["sql"], }, ),
• src/oracle_mcp_server/server.py:631-646 (schema)

  JSON schema defining inputs for execute_query: required 'sql' string, optional 'params' array of strings.
  inputSchema={ "type": "object", "properties": { "sql": { "type": "string", "description": "SQL query to execute (SELECT, DESCRIBE, or EXPLAIN PLAN only)", }, "params": { "type": "array", "items": {"type": "string"}, "description": "Optional parameters for parameterized queries", "default": [], }, }, "required": ["sql"], },
• src/oracle_mcp_server/server.py:769-779 (handler)

  Dispatch handler in MCP call_tool(): extracts 'sql' and 'params' from arguments, invokes QueryExecutor.execute_query, serializes result as TextContent.
  if name == "execute_query": sql = arguments.get("sql") params = arguments.get("params", []) result = await self.executor.execute_query(sql, params) return [ TextContent( type="text", text=json.dumps(result, indent=2, default=str) ) ]