diff --git a/pymongo/_asyncio_task.py b/pymongo/_asyncio_task.py
deleted file mode 100644
index 1b2678dbec..0000000000
--- a/pymongo/_asyncio_task.py
+++ /dev/null
@@ -1,49 +0,0 @@
-# Copyright 2024-present MongoDB, Inc.
-#
-# 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
-#
-# https://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.
-
-"""A custom asyncio.Task that allows checking if a task has been sent a cancellation request.
-Can be removed once we drop Python 3.10 support in favor of asyncio.Task.cancelling."""
-
-from __future__ import annotations
-
-import asyncio
-import sys
-from collections.abc import Coroutine
-from typing import Any, Optional
-
-
-# TODO (https://jira.mongodb.org/browse/PYTHON-4981): Revisit once the underlying cause of the swallowed cancellations is uncovered
-class _Task(asyncio.Task[Any]):
-    def __init__(self, coro: Coroutine[Any, Any, Any], *, name: Optional[str] = None) -> None:
-        super().__init__(coro, name=name)
-        self._cancel_requests = 0
-        asyncio._register_task(self)
-
-    def cancel(self, msg: Optional[str] = None) -> bool:
-        self._cancel_requests += 1
-        return super().cancel(msg=msg)
-
-    def uncancel(self) -> int:
-        if self._cancel_requests > 0:
-            self._cancel_requests -= 1
-        return self._cancel_requests
-
-    def cancelling(self) -> int:
-        return self._cancel_requests
-
-
-def create_task(coro: Coroutine[Any, Any, Any], *, name: Optional[str] = None) -> asyncio.Task[Any]:
-    if sys.version_info >= (3, 11):
-        return asyncio.create_task(coro, name=name)
-    return _Task(coro, name=name)
diff --git a/pymongo/asynchronous/helpers.py b/pymongo/asynchronous/helpers.py
index ba0f483657..ed5b76c554 100644
--- a/pymongo/asynchronous/helpers.py
+++ b/pymongo/asynchronous/helpers.py
@@ -20,7 +20,6 @@
 import builtins
 import functools
 import socket
-import sys
 from typing import (
     Any,
     Callable,
@@ -94,15 +93,5 @@ async def _getaddrinfo(
         return socket.getaddrinfo(host, port, **kwargs)
 
 
-if sys.version_info >= (3, 10):
-    anext = builtins.anext
-    aiter = builtins.aiter
-else:
-
-    async def anext(cls: Any) -> Any:
-        """Compatibility function until we drop 3.9 support: https://docs.python.org/3/library/functions.html#anext."""
-        return await cls.__anext__()
-
-    def aiter(cls: Any) -> Any:
-        """Compatibility function until we drop 3.9 support: https://docs.python.org/3/library/functions.html#anext."""
-        return cls.__aiter__()
+anext = builtins.anext
+aiter = builtins.aiter
diff --git a/pymongo/cursor_shared.py b/pymongo/cursor_shared.py
index 1bde223cb5..2b62b11f04 100644
--- a/pymongo/cursor_shared.py
+++ b/pymongo/cursor_shared.py
@@ -43,7 +43,7 @@
 from pymongo.write_concern import validate_boolean
 
 if TYPE_CHECKING:
-    import sys
+    from typing import Self
 
     from _typeshed import SupportsItems
 
@@ -51,11 +51,6 @@
     from pymongo.message import _OpMsg
     from pymongo.read_preferences import _ServerMode
 
-    if sys.version_info >= (3, 11):
-        from typing import Self
-    else:
-        from typing_extensions import Self
-
 _CURSOR_DOC_FIELDS = {"cursor": {"firstBatch": 1, "nextBatch": 1}}
 
 
diff --git a/pymongo/network_layer.py b/pymongo/network_layer.py
index e368da81a8..f01dd242ec 100644
--- a/pymongo/network_layer.py
+++ b/pymongo/network_layer.py
@@ -33,7 +33,6 @@
 )
 
 from pymongo import _csot, ssl_support
-from pymongo._asyncio_task import create_task
 from pymongo.common import MAX_MESSAGE_SIZE
 from pymongo.compression_support import decompress
 from pymongo.errors import ProtocolError, _OperationCancelled
@@ -711,8 +710,8 @@ async def async_receive_message(
         # timeouts on AWS Lambda and other FaaS environments.
         timeout = max(deadline - time.monotonic(), 0)
 
-    cancellation_task = create_task(_poll_cancellation(conn))
-    read_task = create_task(conn.conn.get_conn.read(request_id, max_message_size))
+    cancellation_task = asyncio.create_task(_poll_cancellation(conn))
+    read_task = asyncio.create_task(conn.conn.get_conn.read(request_id, max_message_size))
     tasks = [read_task, cancellation_task]
     try:
         done, pending = await asyncio.wait(
diff --git a/pymongo/periodic_executor.py b/pymongo/periodic_executor.py
index 867fc33e06..88149115ff 100644
--- a/pymongo/periodic_executor.py
+++ b/pymongo/periodic_executor.py
@@ -24,7 +24,6 @@
 from typing import Any, Optional
 
 from pymongo import _csot, _op_id
-from pymongo._asyncio_task import create_task
 from pymongo.lock import _create_lock
 
 _IS_SYNC = False
@@ -65,9 +64,9 @@ def open(self) -> None:
         self._stopped = False
 
         if self._task is None or (
-            self._task.done() and not self._task.cancelled() and not self._task.cancelling()  # type: ignore[unused-ignore, attr-defined]
+            self._task.done() and not self._task.cancelled() and not self._task.cancelling()
         ):
-            self._task = create_task(self._run(), name=self._name)
+            self._task = asyncio.create_task(self._run(), name=self._name)
 
     def close(self, dummy: Any = None) -> None:
         """Stop. To restart, call open().
@@ -98,7 +97,7 @@ async def _run(self) -> None:
         _csot.reset_all()
         _op_id.reset()
         while not self._stopped:
-            if self._task and self._task.cancelling():  # type: ignore[unused-ignore, attr-defined]
+            if self._task and self._task.cancelling():
                 raise asyncio.CancelledError
             try:
                 if not await self._target():
diff --git a/pymongo/pool_options.py b/pymongo/pool_options.py
index dd95db525e..f20033ad0e 100644
--- a/pymongo/pool_options.py
+++ b/pymongo/pool_options.py
@@ -472,10 +472,7 @@ def _update_metadata(self, driver: DriverInfo) -> None:
             # raises.
             names = metadata["driver"]["name"].split("|")
             versions = metadata["driver"]["version"].split("|")
-            if sys.version_info >= (3, 10):
-                pairs = zip(names, versions, strict=True)
-            else:
-                pairs = zip(names, versions)
+            pairs = zip(names, versions, strict=True)
             if (driver.name, driver.version) in pairs:
                 self.__appended_drivers.add(driver)
 
diff --git a/pymongo/pool_shared.py b/pymongo/pool_shared.py
index 04663f6fc9..3084caedf1 100644
--- a/pymongo/pool_shared.py
+++ b/pymongo/pool_shared.py
@@ -417,11 +417,11 @@ async def _configured_protocol_interface(
 
     host = address[0]
     # asyncio does not support TLS session resumption natively (cpython#79152,
-    # closed without a fix).  On Python 3.11+ SSLProtocol.__init__ calls
-    # wrap_bio() synchronously before the first event-loop yield, so setting
-    # sslobject_class is race-free.  Session injection is skipped on older
-    # Python versions.  (The async path always uses stdlib ssl, never PyOpenSSL.)
-    if ssl_session_cache is not None and sys.version_info >= (3, 11):
+    # closed without a fix).  SSLProtocol.__init__ calls wrap_bio()
+    # synchronously before the first event-loop yield, so setting
+    # sslobject_class is race-free.  (The async path always uses stdlib ssl,
+    # never PyOpenSSL.)
+    if ssl_session_cache is not None:
         session = ssl_session_cache[0]
         if session is not None:
             _session = session
diff --git a/pymongo/synchronous/helpers.py b/pymongo/synchronous/helpers.py
index d0bcb7e32f..3acf43ff5f 100644
--- a/pymongo/synchronous/helpers.py
+++ b/pymongo/synchronous/helpers.py
@@ -20,7 +20,6 @@
 import builtins
 import functools
 import socket
-import sys
 from typing import (
     Any,
     Callable,
@@ -94,15 +93,5 @@ def _getaddrinfo(
         return socket.getaddrinfo(host, port, **kwargs)
 
 
-if sys.version_info >= (3, 10):
-    next = builtins.next
-    iter = builtins.iter
-else:
-
-    def next(cls: Any) -> Any:
-        """Compatibility function until we drop 3.9 support: https://docs.python.org/3/library/functions.html#next."""
-        return cls.__next__()
-
-    def iter(cls: Any) -> Any:
-        """Compatibility function until we drop 3.9 support: https://docs.python.org/3/library/functions.html#next."""
-        return cls.__iter__()
+next = builtins.next
+iter = builtins.iter
diff --git a/test/asynchronous/helpers.py b/test/asynchronous/helpers.py
index 3153eeda52..bbce967767 100644
--- a/test/asynchronous/helpers.py
+++ b/test/asynchronous/helpers.py
@@ -24,7 +24,6 @@
 
 from bson import SON
 from pymongo import common
-from pymongo._asyncio_task import create_task
 from pymongo.read_preferences import ReadPreference
 
 _IS_SYNC = False
@@ -140,7 +139,7 @@ def __init__(self, **kwargs):
     if not _IS_SYNC:
 
         async def start(self):
-            self.task = create_task(self.run(), name=self.name)
+            self.task = asyncio.create_task(self.run(), name=self.name)
 
         async def join(self, timeout: Optional[float] = None):  # type: ignore[override]
             if self.task is not None:
diff --git a/test/asynchronous/test_client.py b/test/asynchronous/test_client.py
index 563d29cbad..46249f762f 100644
--- a/test/asynchronous/test_client.py
+++ b/test/asynchronous/test_client.py
@@ -1678,17 +1678,15 @@ async def test_contextlib(self):
         # pool
         self.assertEqual(1, len((await async_get_pool(client)).conns))
 
-        # contextlib async support was added in Python 3.10
-        if _IS_SYNC or sys.version_info >= (3, 10):
-            async with contextlib.aclosing(client):
-                self.assertEqual("bar", (await client.pymongo_test.coll.find_one())["foo"])
-            with self.assertRaises(InvalidOperation):
-                await client.pymongo_test.coll.find_one()
-            client = await self.async_rs_or_single_client()
-            async with client as client:
-                self.assertEqual("bar", (await client.pymongo_test.coll.find_one())["foo"])
-            with self.assertRaises(InvalidOperation):
-                await client.pymongo_test.coll.find_one()
+        async with contextlib.aclosing(client):
+            self.assertEqual("bar", (await client.pymongo_test.coll.find_one())["foo"])
+        with self.assertRaises(InvalidOperation):
+            await client.pymongo_test.coll.find_one()
+        client = await self.async_rs_or_single_client()
+        async with client as client:
+            self.assertEqual("bar", (await client.pymongo_test.coll.find_one())["foo"])
+        with self.assertRaises(InvalidOperation):
+            await client.pymongo_test.coll.find_one()
 
     @async_client_context.require_sync
     def test_interrupt_signal(self):
diff --git a/test/asynchronous/test_discovery_and_monitoring.py b/test/asynchronous/test_discovery_and_monitoring.py
index 7552054d14..c60cef35c1 100644
--- a/test/asynchronous/test_discovery_and_monitoring.py
+++ b/test/asynchronous/test_discovery_and_monitoring.py
@@ -327,8 +327,6 @@ class TestIgnoreStaleErrors(AsyncIntegrationTest):
         "PYTHON-5861: asyncio.Barrier hangs on macOS ARM64 CI",
     )
     async def test_ignore_stale_connection_errors(self):
-        if not _IS_SYNC and sys.version_info < (3, 11):
-            self.skipTest("Test requires asyncio.Barrier (added in Python 3.11)")
         N_TASKS = 5
         barrier = async_create_barrier(N_TASKS)
         client = await self.async_rs_or_single_client(minPoolSize=N_TASKS)
diff --git a/test/asynchronous/test_encryption.py b/test/asynchronous/test_encryption.py
index e22d185f77..30393956cf 100644
--- a/test/asynchronous/test_encryption.py
+++ b/test/asynchronous/test_encryption.py
@@ -2053,14 +2053,11 @@ async def asyncSetUp(self):
         )
         # Errors when client has no cert, some examples:
         # [SSL: TLSV13_ALERT_CERTIFICATE_REQUIRED] tlsv13 alert certificate required (_ssl.c:2623)
+        # EOF occurred in violation of protocol (_ssl.c:2384)
         self.cert_error = (
             "certificate required|SSL handshake failed|"
-            "KMS connection closed|Connection reset by peer|ECONNRESET|EPIPE"
+            "KMS connection closed|Connection reset by peer|ECONNRESET|EPIPE|EOF"
         )
-        # On Python 3.10+ this error might be:
-        # EOF occurred in violation of protocol (_ssl.c:2384)
-        if sys.version_info[:2] >= (3, 10):
-            self.cert_error += "|EOF"
         # On Windows this error might be:
         # [WinError 10054] An existing connection was forcibly closed by the remote host
         if sys.platform == "win32":
diff --git a/test/asynchronous/test_locks.py b/test/asynchronous/test_locks.py
index 54c31cbe35..4d207b509b 100644
--- a/test/asynchronous/test_locks.py
+++ b/test/asynchronous/test_locks.py
@@ -299,9 +299,6 @@ async def test_timeout_in_block(self):
                 with self.assertRaises(asyncio.TimeoutError):
                     await asyncio.wait_for(condition.wait(), timeout=0.5)
 
-        @unittest.skipIf(
-            sys.version_info < (3, 11), "raising the same cancelled error requires Python>=3.11"
-        )
         async def test_cancelled_error_wakeup(self):
             # Test that a cancelled error, received when awaiting wakeup,
             # will be re-raised un-modified.
@@ -328,9 +325,6 @@ async def func():
             # originally raised.
             self.assertIs(err.exception, raised)
 
-        @unittest.skipIf(
-            sys.version_info < (3, 11), "raising the same cancelled error requires Python>=3.11"
-        )
         async def test_cancelled_error_re_aquire(self):
             # Test that a cancelled error, received when re-aquiring lock,
             # will be re-raised un-modified.
@@ -363,7 +357,6 @@ async def func():
             # originally raised.
             self.assertIs(err.exception, raised)
 
-        @unittest.skipIf(sys.version_info < (3, 11), "asyncio.timeout requires Python>=3.11")
         async def test_cancelled_wakeup(self):
             # Test that a task cancelled at the "same" time as it is woken
             # up as part of a Condition.notify() does not result in a lost wakeup.
@@ -409,7 +402,6 @@ async def consumer():
                 condition.notify_all()
             await c[1]
 
-        @unittest.skipIf(sys.version_info < (3, 11), "asyncio.timeout requires Python>=3.11")
         async def test_cancelled_wakeup_relock(self):
             # Test that a task cancelled at the "same" time as it is woken
             # up as part of a Condition.notify() does not result in a lost wakeup.
diff --git a/test/asynchronous/test_retryable_writes.py b/test/asynchronous/test_retryable_writes.py
index bcceae1311..b9fdb439aa 100644
--- a/test/asynchronous/test_retryable_writes.py
+++ b/test/asynchronous/test_retryable_writes.py
@@ -23,7 +23,6 @@
 import threading
 from unittest import mock
 
-from pymongo._asyncio_task import create_task
 from pymongo.common import MAX_ADAPTIVE_RETRIES
 from test.asynchronous.utils import async_set_fail_point, flaky
 
@@ -104,7 +103,7 @@ def succeeded(self, event: CommandSucceededEvent) -> None:
             else:
                 # succeeded() cannot await, so the fail point may be configured after
                 # the driver dispatches the retry it is meant to fail.
-                self._task = create_task(async_client_context.client.admin.command(cmd))  # type: ignore[arg-type]
+                self._task = asyncio.create_task(async_client_context.client.admin.command(cmd))  # type: ignore[arg-type]
                 self._task.add_done_callback(self._store_fail_point_exc)
 
 
diff --git a/test/asynchronous/test_ssl.py b/test/asynchronous/test_ssl.py
index 81a7cb9630..1ec39c5e09 100644
--- a/test/asynchronous/test_ssl.py
+++ b/test/asynchronous/test_ssl.py
@@ -189,10 +189,7 @@ class FakeSSLSock:
 
         self.assertIs(_get_ssl_session(FakeSSLSock()), fake_session)
 
-    @unittest.skipUnless(
-        not _IS_SYNC and sys.version_info >= (3, 11),
-        "Tests async sslobject_class injection (Python 3.11+ only)",
-    )
+    @unittest.skipUnless(not _IS_SYNC, "Tests async sslobject_class injection only")
     def test_async_tls_session_injected_via_sslobject_class(self):
         """On Python 3.11+, a cached session is injected by setting sslobject_class."""
         fake_session = object()
@@ -260,10 +257,7 @@ async def run():
         asyncio.run(run())
         self.assertIs(cache[0], fake_session)
 
-    @unittest.skipUnless(
-        not _IS_SYNC and sys.version_info >= (3, 11),
-        "Tests async session injection on Python 3.11+",
-    )
+    @unittest.skipUnless(not _IS_SYNC, "Tests async session injection only")
     def test_async_configured_protocol_injects_session_via_sslobject_class(self):
         """When the cache has a session, sslobject_class is set and its __init__ body runs."""
         initial_session = object()
diff --git a/test/helpers.py b/test/helpers.py
index 5b9afa192e..6814326d38 100644
--- a/test/helpers.py
+++ b/test/helpers.py
@@ -24,7 +24,6 @@
 
 from bson import SON
 from pymongo import common
-from pymongo._asyncio_task import create_task
 from pymongo.read_preferences import ReadPreference
 
 _IS_SYNC = True
@@ -140,7 +139,7 @@ def __init__(self, **kwargs):
     if not _IS_SYNC:
 
         def start(self):
-            self.task = create_task(self.run(), name=self.name)
+            self.task = asyncio.create_task(self.run(), name=self.name)
 
         def join(self, timeout: Optional[float] = None):  # type: ignore[override]
             if self.task is not None:
diff --git a/test/test_client.py b/test/test_client.py
index 0c8d0e7595..a2640def28 100644
--- a/test/test_client.py
+++ b/test/test_client.py
@@ -1633,17 +1633,15 @@ def test_contextlib(self):
         # pool
         self.assertEqual(1, len((get_pool(client)).conns))
 
-        # contextlib async support was added in Python 3.10
-        if _IS_SYNC or sys.version_info >= (3, 10):
-            with contextlib.closing(client):
-                self.assertEqual("bar", (client.pymongo_test.coll.find_one())["foo"])
-            with self.assertRaises(InvalidOperation):
-                client.pymongo_test.coll.find_one()
-            client = self.rs_or_single_client()
-            with client as client:
-                self.assertEqual("bar", (client.pymongo_test.coll.find_one())["foo"])
-            with self.assertRaises(InvalidOperation):
-                client.pymongo_test.coll.find_one()
+        with contextlib.closing(client):
+            self.assertEqual("bar", (client.pymongo_test.coll.find_one())["foo"])
+        with self.assertRaises(InvalidOperation):
+            client.pymongo_test.coll.find_one()
+        client = self.rs_or_single_client()
+        with client as client:
+            self.assertEqual("bar", (client.pymongo_test.coll.find_one())["foo"])
+        with self.assertRaises(InvalidOperation):
+            client.pymongo_test.coll.find_one()
 
     @client_context.require_sync
     def test_interrupt_signal(self):
diff --git a/test/test_discovery_and_monitoring.py b/test/test_discovery_and_monitoring.py
index 2086f56a2b..3bec615274 100644
--- a/test/test_discovery_and_monitoring.py
+++ b/test/test_discovery_and_monitoring.py
@@ -327,8 +327,6 @@ class TestIgnoreStaleErrors(IntegrationTest):
         "PYTHON-5861: asyncio.Barrier hangs on macOS ARM64 CI",
     )
     def test_ignore_stale_connection_errors(self):
-        if not _IS_SYNC and sys.version_info < (3, 11):
-            self.skipTest("Test requires asyncio.Barrier (added in Python 3.11)")
         N_TASKS = 5
         barrier = create_barrier(N_TASKS)
         client = self.rs_or_single_client(minPoolSize=N_TASKS)
diff --git a/test/test_encryption.py b/test/test_encryption.py
index 2f9ed382ca..5e07c011fa 100644
--- a/test/test_encryption.py
+++ b/test/test_encryption.py
@@ -2045,14 +2045,11 @@ def setUp(self):
         )
         # Errors when client has no cert, some examples:
         # [SSL: TLSV13_ALERT_CERTIFICATE_REQUIRED] tlsv13 alert certificate required (_ssl.c:2623)
+        # EOF occurred in violation of protocol (_ssl.c:2384)
         self.cert_error = (
             "certificate required|SSL handshake failed|"
-            "KMS connection closed|Connection reset by peer|ECONNRESET|EPIPE"
+            "KMS connection closed|Connection reset by peer|ECONNRESET|EPIPE|EOF"
         )
-        # On Python 3.10+ this error might be:
-        # EOF occurred in violation of protocol (_ssl.c:2384)
-        if sys.version_info[:2] >= (3, 10):
-            self.cert_error += "|EOF"
         # On Windows this error might be:
         # [WinError 10054] An existing connection was forcibly closed by the remote host
         if sys.platform == "win32":
diff --git a/test/test_retryable_writes.py b/test/test_retryable_writes.py
index 5a34bc90e5..a82a1c5731 100644
--- a/test/test_retryable_writes.py
+++ b/test/test_retryable_writes.py
@@ -23,7 +23,6 @@
 import threading
 from unittest import mock
 
-from pymongo._asyncio_task import create_task
 from pymongo.common import MAX_ADAPTIVE_RETRIES
 from test.utils import flaky, set_fail_point
 
@@ -104,7 +103,7 @@ def succeeded(self, event: CommandSucceededEvent) -> None:
             else:
                 # succeeded() cannot await, so the fail point may be configured after
                 # the driver dispatches the retry it is meant to fail.
-                self._task = create_task(client_context.client.admin.command(cmd))  # type: ignore[arg-type]
+                self._task = asyncio.create_task(client_context.client.admin.command(cmd))  # type: ignore[arg-type]
                 self._task.add_done_callback(self._store_fail_point_exc)
 
 
diff --git a/test/test_ssl.py b/test/test_ssl.py
index c614f016e7..9e76de2cf8 100644
--- a/test/test_ssl.py
+++ b/test/test_ssl.py
@@ -188,10 +188,7 @@ class FakeSSLSock:
 
         self.assertIs(_get_ssl_session(FakeSSLSock()), fake_session)
 
-    @unittest.skipUnless(
-        not _IS_SYNC and sys.version_info >= (3, 11),
-        "Tests async sslobject_class injection (Python 3.11+ only)",
-    )
+    @unittest.skipUnless(not _IS_SYNC, "Tests async sslobject_class injection only")
     def test_async_tls_session_injected_via_sslobject_class(self):
         """On Python 3.11+, a cached session is injected by setting sslobject_class."""
         fake_session = object()
@@ -257,10 +254,7 @@ def run():
         asyncio.run(run())
         self.assertIs(cache[0], fake_session)
 
-    @unittest.skipUnless(
-        not _IS_SYNC and sys.version_info >= (3, 11),
-        "Tests async session injection on Python 3.11+",
-    )
+    @unittest.skipUnless(not _IS_SYNC, "Tests async session injection only")
     def test_async_configured_protocol_injects_session_via_sslobject_class(self):
         """When the cache has a session, sslobject_class is set and its __init__ body runs."""
         initial_session = object()