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)
          )
      ]