diff --git a/LICENSE b/LICENSE
index 2ba7df1a32b1e..e6c5aeb5ec955 100644
--- a/LICENSE
+++ b/LICENSE
@@ -289,7 +289,7 @@ Code in python/ray/_private/prometheus_exporter.py is adapted from https://githu
 # limitations under the License.
 
 --------------------------------------------------------------------------------
-Code in python/ray/tests/modin/test_modin and 
+Code in python/ray/tests/modin/test_modin and
 python/ray/tests/modin/modin_test_utils adapted from:
 - http://github.com/modin-project/modin/master/modin/pandas/test/test_general.py
 - http://github.com/modin-project/modin/master/modin/pandas/test/utils.py
@@ -309,7 +309,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 
 --------------------------------------------------------------------------------
-Code in src/ray/util/logging.h is adapted from 
+Code in src/ray/util/logging.h is adapted from
 https://github.com/google/glog/blob/master/src/glog/logging.h.in
 
 Copyright (c) 2008, Google Inc.
@@ -342,7 +342,7 @@ THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 --------------------------------------------------------------------------------
-Code in python/ray/_private/runtime_env/conda_utils.py is adapted from 
+Code in python/ray/_private/runtime_env/conda_utils.py is adapted from
 https://github.com/mlflow/mlflow/blob/master/mlflow/utils/conda.py
 
 Copyright (c) 2018, Databricks, Inc.
@@ -384,24 +384,6 @@ LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 THE SOFTWARE.
 
---------------------------------------------------------------------------------
-Code in python/ray/_private/async_compat.py is adapted from
-https://github.com/python-trio/async_generator/blob/master/async_generator/_util.py
-
-Copyright (c) 2022, Nathaniel J. Smith
-
-Licensed under the Apache License, Version 2.0 (the "License");
-you may not use this file except in compliance with the License.
-You may obtain a copy of the License at
-
-   http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-
 ---------------------------------------------------------------------------------------------------------------
 Code in python/ray/_private/thirdparty/tabulate/tabulate.py is adapted from https://github.com/astanin/python-tabulate/blob/4892c6e9a79638c7897ccea68b602040da9cc7a7/tabulate.py
 
diff --git a/dashboard/modules/runtime_env/runtime_env_agent.py b/dashboard/modules/runtime_env/runtime_env_agent.py
index 1ebbbfd2226f3..babaa97eb0d41 100644
--- a/dashboard/modules/runtime_env/runtime_env_agent.py
+++ b/dashboard/modules/runtime_env/runtime_env_agent.py
@@ -14,7 +14,6 @@
 import ray.dashboard.consts as dashboard_consts
 import ray.dashboard.modules.runtime_env.runtime_env_consts as runtime_env_consts
 import ray.dashboard.utils as dashboard_utils
-from ray._private.async_compat import create_task
 from ray._private.ray_logging import setup_component_logger
 from ray._private.runtime_env.conda import CondaPlugin
 from ray._private.runtime_env.container import ContainerManager
@@ -344,15 +343,10 @@ async def _create_runtime_env_with_retry(
             error_message = None
             for _ in range(runtime_env_consts.RUNTIME_ENV_RETRY_TIMES):
                 try:
-                    # python 3.6 requires the type of input is `Future`,
-                    # python 3.7+ only requires the type of input is `Awaitable`
-                    # TODO(Catch-Bull): remove create_task when ray drop python 3.6
-                    runtime_env_setup_task = create_task(
-                        _setup_runtime_env(
-                            runtime_env,
-                            serialized_env,
-                            request.serialized_allocated_resource_instances,
-                        )
+                    runtime_env_setup_task = _setup_runtime_env(
+                        runtime_env,
+                        serialized_env,
+                        request.serialized_allocated_resource_instances,
                     )
                     runtime_env_context = await asyncio.wait_for(
                         runtime_env_setup_task, timeout=setup_timeout_seconds
diff --git a/python/ray/_private/async_compat.py b/python/ray/_private/async_compat.py
index 06b0d47f9b8dd..b1ecccf2590ec 100644
--- a/python/ray/_private/async_compat.py
+++ b/python/ray/_private/async_compat.py
@@ -11,20 +11,6 @@
     uvloop = None
 
 
-try:
-    # This function has been added in Python 3.7. Prior to Python 3.7,
-    # the low-level asyncio.ensure_future() function can be used instead.
-    from asyncio import create_task  # noqa: F401
-except ImportError:
-    from asyncio import ensure_future as create_task  # noqa: F401
-
-
-try:
-    from asyncio import get_running_loop  # noqa: F401
-except ImportError:
-    from asyncio import _get_running_loop as get_running_loop  # noqa: F401
-
-
 def get_new_event_loop():
     """Construct a new event loop. Ray will use uvloop if it exists"""
     if uvloop:
@@ -43,116 +29,3 @@ async def wrapper(*args, **kwargs):
         return func(*args, **kwargs)
 
     return wrapper
-
-
-try:
-    from contextlib import asynccontextmanager
-except ImportError:
-    # Copy from https://github.com/python-trio/async_generator
-    # for compatible with Python 3.6
-    import sys
-    from functools import wraps
-    from inspect import isasyncgenfunction
-
-    class _aclosing:
-        def __init__(self, aiter):
-            self._aiter = aiter
-
-        async def __aenter__(self):
-            return self._aiter
-
-        async def __aexit__(self, *args):
-            await self._aiter.aclose()
-
-    # Very much derived from the one in contextlib, by copy/pasting and then
-    # asyncifying everything. (Also I dropped the obscure support for using
-    # context managers as function decorators. It could be re-added; I just
-    # couldn't be bothered.)
-    # So this is a derivative work licensed under the PSF License, which requires
-    # the following notice:
-    #
-    # Copyright © 2001-2017 Python Software Foundation; All Rights Reserved
-    class _AsyncGeneratorContextManager:
-        def __init__(self, func, args, kwds):
-            self._func_name = func.__name__
-            self._agen = func(*args, **kwds).__aiter__()
-
-        async def __aenter__(self):
-            try:
-                return await self._agen.asend(None)
-            except StopAsyncIteration:
-                raise RuntimeError("async generator didn't yield") from None
-
-        async def __aexit__(self, type, value, traceback):
-            async with _aclosing(self._agen):
-                if type is None:
-                    try:
-                        await self._agen.asend(None)
-                    except StopAsyncIteration:
-                        return False
-                    else:
-                        raise RuntimeError("async generator didn't stop")
-                else:
-                    # It used to be possible to have type != None, value == None:
-                    #    https://bugs.python.org/issue1705170
-                    # but AFAICT this can't happen anymore.
-                    assert value is not None
-                    try:
-                        await self._agen.athrow(type, value, traceback)
-                        raise RuntimeError("async generator didn't stop after athrow()")
-                    except StopAsyncIteration as exc:
-                        # Suppress StopIteration *unless* it's the same exception
-                        # that was passed to throw(). This prevents a
-                        # StopIteration raised inside the "with" statement from
-                        # being suppressed.
-                        return exc is not value
-                    except RuntimeError as exc:
-                        # Don't re-raise the passed in exception. (issue27112)
-                        if exc is value:
-                            return False
-                        # Likewise, avoid suppressing if a StopIteration exception
-                        # was passed to throw() and later wrapped into a
-                        # RuntimeError (see PEP 479).
-                        if (
-                            isinstance(value, (StopIteration, StopAsyncIteration))
-                            and exc.__cause__ is value
-                        ):
-                            return False
-                        raise
-                    except:  # noqa: E722
-                        # only re-raise if it's *not* the exception that was
-                        # passed to throw(), because __exit__() must not raise an
-                        # exception unless __exit__() itself failed. But throw()
-                        # has to raise the exception to signal propagation, so
-                        # this fixes the impedance mismatch between the throw()
-                        # protocol and the __exit__() protocol.
-                        #
-                        if sys.exc_info()[1] is value:
-                            return False
-                        raise
-
-        def __enter__(self):
-            raise RuntimeError(
-                "use 'async with {func_name}(...)', not 'with {func_name}(...)'".format(
-                    func_name=self._func_name
-                )
-            )
-
-        def __exit__(self):  # pragma: no cover
-            assert False, """Never called, but should be defined"""
-
-    def asynccontextmanager(func):
-        """Like @contextmanager, but async."""
-        if not isasyncgenfunction(func):
-            raise TypeError(
-                "must be an async generator (native or from async_generator; "
-                "if using @async_generator then @acontextmanager must be on top."
-            )
-
-        @wraps(func)
-        def helper(*args, **kwds):
-            return _AsyncGeneratorContextManager(func, args, kwds)
-
-        # A hint for sphinxcontrib-trio:
-        helper.__returns_acontextmanager__ = True
-        return helper
diff --git a/python/ray/_private/function_manager.py b/python/ray/_private/function_manager.py
index 2db314ce1aabb..9dd9173cf2654 100644
--- a/python/ray/_private/function_manager.py
+++ b/python/ray/_private/function_manager.py
@@ -5,7 +5,6 @@
 import json
 import logging
 import os
-import sys
 import threading
 import time
 import traceback
@@ -128,10 +127,7 @@ def compute_collision_identifier(self, function_or_class):
         import io
 
         string_file = io.StringIO()
-        if sys.version_info[1] >= 7:
-            dis.dis(function_or_class, file=string_file, depth=2)
-        else:
-            dis.dis(function_or_class, file=string_file)
+        dis.dis(function_or_class, file=string_file, depth=2)
         collision_identifier = function_or_class.__name__ + ":" + string_file.getvalue()
 
         # Return a hash of the identifier in case it is too large.
diff --git a/python/ray/_private/runtime_env/pip.py b/python/ray/_private/runtime_env/pip.py
index a25ecd28a9722..b70f165d5b545 100644
--- a/python/ray/_private/runtime_env/pip.py
+++ b/python/ray/_private/runtime_env/pip.py
@@ -7,8 +7,9 @@
 import sys
 import tempfile
 from typing import Dict, List, Optional, Tuple
+from contextlib import asynccontextmanager
+from asyncio import create_task, get_running_loop
 
-from ray._private.async_compat import asynccontextmanager, create_task, get_running_loop
 from ray._private.runtime_env.context import RuntimeEnvContext
 from ray._private.runtime_env.packaging import Protocol, parse_uri
 from ray._private.runtime_env.plugin import RuntimeEnvPlugin
diff --git a/python/ray/_private/utils.py b/python/ray/_private/utils.py
index 05e85d2719192..d35be4353bd70 100644
--- a/python/ray/_private/utils.py
+++ b/python/ray/_private/utils.py
@@ -1800,9 +1800,7 @@ def create_if_main_thread(cls) -> contextlib.AbstractContextManager:
         if threading.current_thread() == threading.main_thread():
             return cls()
         else:
-            # TODO(Clark): Use contextlib.nullcontext() once Python 3.6 support is
-            # dropped.
-            return contextlib.suppress()
+            return contextlib.nullcontext()
 
     def _set_task_cancelled(self, signum, frame):
         """SIGINT handler that defers the signal."""
diff --git a/python/ray/data/__init__.py b/python/ray/data/__init__.py
index 8e813ef5a5301..d6ef16526f9d9 100644
--- a/python/ray/data/__init__.py
+++ b/python/ray/data/__init__.py
@@ -1,10 +1,6 @@
-import sys
-
 # Short term workaround for https://github.com/ray-project/ray/issues/32435
 # Datastream has a hard dependency on pandas, so it doesn't need to be delayed.
-# ray.data import is still eager for all ray imports for Python 3.6:
-if sys.version_info >= (3, 7):
-    import pandas  # noqa
+import pandas  # noqa
 
 from ray.data._internal.compute import ActorPoolStrategy
 from ray.data._internal.progress_bar import set_progress_bars
diff --git a/python/ray/data/_internal/block_batching/block_batching.py b/python/ray/data/_internal/block_batching/block_batching.py
index c6e655aca66e3..6fc4990f3674a 100644
--- a/python/ray/data/_internal/block_batching/block_batching.py
+++ b/python/ray/data/_internal/block_batching/block_batching.py
@@ -1,7 +1,7 @@
 import collections
 import itertools
-import sys
 from typing import Any, Callable, Iterator, Optional, TypeVar, Union
+from contextlib import nullcontext
 
 import ray
 from ray.data._internal.block_batching.interfaces import BlockPrefetcher
@@ -22,15 +22,6 @@
 
 T = TypeVar("T")
 
-if sys.version_info >= (3, 7):
-    from contextlib import nullcontext
-else:
-    from contextlib import contextmanager
-
-    @contextmanager
-    def nullcontext(enter_result=None):
-        yield enter_result
-
 
 def batch_block_refs(
     block_refs: Iterator[ObjectRef[Block]],
diff --git a/python/ray/data/_internal/block_batching/iter_batches.py b/python/ray/data/_internal/block_batching/iter_batches.py
index 9e59583d6abe0..e3c2e8b56e066 100644
--- a/python/ray/data/_internal/block_batching/iter_batches.py
+++ b/python/ray/data/_internal/block_batching/iter_batches.py
@@ -1,5 +1,4 @@
 import collections
-import sys
 from typing import Any, Callable, Dict, Iterator, Optional, Tuple
 
 import ray
@@ -22,15 +21,7 @@
 from ray.data._internal.memory_tracing import trace_deallocation
 from ray.data._internal.stats import DatastreamStats
 from ray.data.context import DataContext
-
-if sys.version_info >= (3, 7):
-    from contextlib import nullcontext
-else:
-    from contextlib import contextmanager
-
-    @contextmanager
-    def nullcontext(enter_result=None):
-        yield enter_result
+from contextlib import nullcontext
 
 
 def iter_batches(
diff --git a/python/ray/data/_internal/block_batching/util.py b/python/ray/data/_internal/block_batching/util.py
index c642f6c2ac638..a82b46ae43a73 100644
--- a/python/ray/data/_internal/block_batching/util.py
+++ b/python/ray/data/_internal/block_batching/util.py
@@ -1,8 +1,8 @@
 import logging
 import queue
 import threading
-import sys
 from typing import Any, Callable, Iterator, List, Optional, Tuple, TypeVar, Union
+from contextlib import nullcontext
 
 import ray
 from ray.types import ObjectRef
@@ -22,15 +22,6 @@
 
 logger = logging.getLogger(__name__)
 
-if sys.version_info >= (3, 7):
-    from contextlib import nullcontext
-else:
-    from contextlib import contextmanager
-
-    @contextmanager
-    def nullcontext(enter_result=None):
-        yield enter_result
-
 
 def _calculate_ref_hits(refs: List[ObjectRef[Any]]) -> Tuple[int, int, int]:
     """Given a list of object references, returns how many are already on the local
diff --git a/python/ray/serve/benchmarks/handle.py b/python/ray/serve/benchmarks/handle.py
index cd078fde0d582..b925ce9eabbfd 100644
--- a/python/ray/serve/benchmarks/handle.py
+++ b/python/ray/serve/benchmarks/handle.py
@@ -94,5 +94,4 @@ async def main():
                 await run_test(num_replicas, num_forwarders, sync)
 
 
-# TODO(rickyx): use asyncio.run after deprecating 3.6
-asyncio.new_event_loop().run_until_complete(main())
+asyncio.run(main())
diff --git a/python/ray/tests/modin/test_modin.py b/python/ray/tests/modin/test_modin.py
index 5bbe52613a0c2..aec9f98ee4357 100644
--- a/python/ray/tests/modin/test_modin.py
+++ b/python/ray/tests/modin/test_modin.py
@@ -24,16 +24,14 @@
 import ray
 from ray.tests.conftest import start_cluster  # noqa F401
 
-modin_compatible_version = sys.version_info >= (3, 7, 0)
 modin_installed = True
 
-if modin_compatible_version:
-    try:
-        import modin  # noqa: F401
-    except ModuleNotFoundError:
-        modin_installed = False
+try:
+    import modin  # noqa: F401
+except ModuleNotFoundError:
+    modin_installed = False
 
-skip = not modin_compatible_version or not modin_installed
+skip = not modin_installed
 
 # These tests are written for versions of Modin that require python 3.7+
 pytestmark = pytest.mark.skipif(skip, reason="Outdated or missing Modin dependency")
diff --git a/python/ray/tests/test_gcs_pubsub.py b/python/ray/tests/test_gcs_pubsub.py
index f14d7be0fca4f..b9a4eddee7a40 100644
--- a/python/ray/tests/test_gcs_pubsub.py
+++ b/python/ray/tests/test_gcs_pubsub.py
@@ -142,9 +142,6 @@ async def test_aio_publish_and_subscribe_resource_usage(ray_start_regular):
     await subscriber.close()
 
 
-@pytest.mark.skipif(
-    sys.version_info < (3, 7, 0), reason="no asyncio.all_tasks in py3.6"
-)
 @pytest.mark.asyncio
 async def test_aio_poll_no_leaks(ray_start_regular):
     """Test that polling doesn't leak memory."""
diff --git a/python/ray/tests/test_output.py b/python/ray/tests/test_output.py
index 6310c3517e005..9299cb4e5d4fd 100644
--- a/python/ray/tests/test_output.py
+++ b/python/ray/tests/test_output.py
@@ -362,7 +362,6 @@ def test_core_worker_error_message():
 
 
 @pytest.mark.skipif(sys.platform == "win32", reason="Failing on Windows.")
-@pytest.mark.skipif(sys.version_info < (3, 7), reason="requires python3.7+")
 def test_disable_driver_logs_breakpoint():
     script = """
 import time
diff --git a/python/ray/tests/test_serialization.py b/python/ray/tests/test_serialization.py
index 43a7f02a29cec..084ff8b5cf376 100644
--- a/python/ray/tests/test_serialization.py
+++ b/python/ray/tests/test_serialization.py
@@ -6,6 +6,7 @@
 import string
 import sys
 import weakref
+from dataclasses import make_dataclass
 
 import numpy as np
 import pytest
@@ -245,30 +246,26 @@ class CustomError(Exception):
         NamedTupleExample(1, 1.0, "hi", np.zeros([3, 5]), [1, 2, 3]),
     ]
 
-    # Test dataclasses in Python 3.7.
-    if sys.version_info >= (3, 7):
-        from dataclasses import make_dataclass
+    DataClass0 = make_dataclass("DataClass0", [("number", int)])
 
-        DataClass0 = make_dataclass("DataClass0", [("number", int)])
+    CUSTOM_OBJECTS.append(DataClass0(number=3))
 
-        CUSTOM_OBJECTS.append(DataClass0(number=3))
-
-        class CustomClass:
-            def __init__(self, value):
-                self.value = value
+    class CustomClass:
+        def __init__(self, value):
+            self.value = value
 
-        DataClass1 = make_dataclass("DataClass1", [("custom", CustomClass)])
+    DataClass1 = make_dataclass("DataClass1", [("custom", CustomClass)])
 
-        class DataClass2(DataClass1):
-            @classmethod
-            def from_custom(cls, data):
-                custom = CustomClass(data)
-                return cls(custom)
+    class DataClass2(DataClass1):
+        @classmethod
+        def from_custom(cls, data):
+            custom = CustomClass(data)
+            return cls(custom)
 
-            def __reduce__(self):
-                return (self.from_custom, (self.custom.value,))
+        def __reduce__(self):
+            return (self.from_custom, (self.custom.value,))
 
-        CUSTOM_OBJECTS.append(DataClass2(custom=CustomClass(43)))
+    CUSTOM_OBJECTS.append(DataClass2(custom=CustomClass(43)))
 
     BASE_OBJECTS = PRIMITIVE_OBJECTS + COMPLEX_OBJECTS + CUSTOM_OBJECTS
 
diff --git a/python/ray/train/huggingface/accelerate/_accelerate_utils.py b/python/ray/train/huggingface/accelerate/_accelerate_utils.py
index 2b6e29b869c82..64a362bc38678 100644
--- a/python/ray/train/huggingface/accelerate/_accelerate_utils.py
+++ b/python/ray/train/huggingface/accelerate/_accelerate_utils.py
@@ -16,21 +16,13 @@
 
 import logging
 import os
-import sys
 from argparse import Namespace
 from typing import Optional, Tuple, Union
 import tempfile
 from pathlib import Path
 
 
-if sys.version_info >= (3, 7):
-    from contextlib import nullcontext
-else:
-    from contextlib import contextmanager
-
-    @contextmanager
-    def nullcontext(enter_result=None):
-        yield enter_result
+from contextlib import nullcontext
 
 
 try:
diff --git a/python/ray/util/client/dataclient.py b/python/ray/util/client/dataclient.py
index cfd6907837024..449c765343817 100644
--- a/python/ray/util/client/dataclient.py
+++ b/python/ray/util/client/dataclient.py
@@ -4,7 +4,6 @@
 import math
 import logging
 import queue
-import sys
 import threading
 import warnings
 import grpc
@@ -414,7 +413,7 @@ def _reconnect_channel(self) -> None:
     # Use SimpleQueue to avoid deadlocks when appending to queue from __del__()
     @staticmethod
     def _create_queue():
-        return queue.Queue() if sys.version_info < (3, 7) else queue.SimpleQueue()
+        return queue.SimpleQueue()
 
     def close(self) -> None:
         thread = None
diff --git a/python/ray/util/rpdb.py b/python/ray/util/rpdb.py
index 825c0acf2a72b..63144706032a3 100644
--- a/python/ray/util/rpdb.py
+++ b/python/ray/util/rpdb.py
@@ -24,7 +24,6 @@
 from ray.experimental.internal_kv import _internal_kv_del, _internal_kv_put
 from ray.util.annotations import DeveloperAPI
 
-PY3 = sys.version_info[0] == 3
 log = logging.getLogger(__name__)
 
 
diff --git a/python/setup.py b/python/setup.py
index 10815416af5a4..6cb8a41be1d5a 100644
--- a/python/setup.py
+++ b/python/setup.py
@@ -774,7 +774,6 @@ def has_ext_modules(self):
         "reinforcement-learning deep-learning serving python"
     ),
     classifiers=[
-        "Programming Language :: Python :: 3.6",
         "Programming Language :: Python :: 3.7",
         "Programming Language :: Python :: 3.8",
         "Programming Language :: Python :: 3.9",