diff --git a/ldclient/impl/aio/__init__.py b/ldclient/impl/aio/__init__.py
new file mode 100644
index 00000000..1e34a859
--- /dev/null
+++ b/ldclient/impl/aio/__init__.py
@@ -0,0 +1,18 @@
+"""
+Async support classes backing the async data source, event processor, and data
+system. These have no sync twin: the sync code uses the equivalent stdlib/SDK
+primitives (``threading``, ``queue.Queue``, ``RepeatingTask``,
+``FixedThreadPool``, urllib3) directly.
+
+``aio.concurrency`` holds the async concurrency primitives (``AsyncEvent``,
+``AsyncQueue``, ``AsyncTaskRunner``, ``AsyncRepeatingTask``, ``AsyncWorkerPool``,
+etc.) that wrap fiddly asyncio plumbing the async code would otherwise inline
+repeatedly.
+
+``aio.transport`` wraps an aiohttp ``ClientSession`` behind a
+``TransportResponse`` (see ``aio.transport_types``) so async data-source callers
+can inspect a response after the request context has closed and so the
+SSL/session setup is written once. The sync side talks to urllib3 directly.
+
+These are covered by ``ldclient/testing/test_aio.py``.
+"""
diff --git a/ldclient/impl/aio/concurrency.py b/ldclient/impl/aio/concurrency.py
new file mode 100644
index 00000000..f9529ff4
--- /dev/null
+++ b/ldclient/impl/aio/concurrency.py
@@ -0,0 +1,305 @@
+"""
+Async concurrency primitives used by the async data source, event processor, and
+data system. Each wraps a piece of fiddly asyncio plumbing (timeout-aware waits,
+queue exception normalization, an interval-from-start repeating task, a bounded
+task pool) that callers would otherwise inline repeatedly. The sync code uses the
+equivalent stdlib/SDK primitives (``threading.Event``/``Lock``, ``queue.Queue``,
+``RepeatingTask``, ``FixedThreadPool``) directly, so these have no sync twin.
+"""
+
+import asyncio
+import inspect
+import time
+from queue import Empty as QueueEmpty  # noqa: F401  (shared timeout exception)
+from queue import Full as QueueFull  # noqa: F401  (shared capacity exception)
+from typing import Any, Callable, Optional, Set
+
+from ldclient.impl.util import log
+
+
+class AsyncEvent:
+    """Wraps ``asyncio.Event``, adding a ``wait(timeout)`` that returns False on
+    timeout instead of raising, to match ``threading.Event.wait``."""
+
+    def __init__(self):
+        self._event = asyncio.Event()
+
+    def set(self) -> None:
+        self._event.set()
+
+    def clear(self) -> None:
+        self._event.clear()
+
+    def is_set(self) -> bool:
+        return self._event.is_set()
+
+    async def wait(self, timeout: Optional[float] = None) -> bool:
+        if timeout is None:
+            await self._event.wait()
+            return True
+        try:
+            await asyncio.wait_for(self._event.wait(), timeout)
+            return True
+        except asyncio.TimeoutError:
+            return False
+
+
+class AsyncLock:
+    """Wraps ``asyncio.Lock`` as an async context manager."""
+
+    def __init__(self):
+        self._lock = asyncio.Lock()
+
+    async def __aenter__(self):
+        await self._lock.acquire()
+        return self
+
+    async def __aexit__(self, exc_type, exc_value, traceback):
+        self._lock.release()
+        return False
+
+    def locked(self) -> bool:
+        return self._lock.locked()
+
+
+class AsyncQueue:
+    """Wraps ``asyncio.Queue`` with a ``get(timeout)`` that raises the shared
+    ``QueueEmpty`` on timeout, and a ``put_nowait`` that raises the shared
+    ``QueueFull`` when a bounded queue is at capacity."""
+
+    def __init__(self, maxsize: int = 0):
+        self._queue: asyncio.Queue = asyncio.Queue(maxsize=maxsize)
+
+    async def put(self, item: Any) -> None:
+        await self._queue.put(item)
+
+    def put_nowait(self, item: Any) -> None:
+        try:
+            self._queue.put_nowait(item)
+        except asyncio.QueueFull:
+            raise QueueFull() from None
+
+    async def get(self, timeout: Optional[float] = None, block: bool = True) -> Any:
+        if not block:
+            try:
+                return self._queue.get_nowait()
+            except asyncio.QueueEmpty:
+                raise QueueEmpty() from None
+        if timeout is None:
+            return await self._queue.get()
+        try:
+            return await asyncio.wait_for(self._queue.get(), timeout)
+        except asyncio.TimeoutError:
+            raise QueueEmpty()
+
+    def empty(self) -> bool:
+        return self._queue.empty()
+
+
+# The handle type returned by spawn_handle.
+TaskHandle = asyncio.Task
+
+
+def _log_task_exception(task: asyncio.Task) -> None:
+    if not task.cancelled() and task.exception() is not None:
+        log.error("Unhandled exception in background task", exc_info=task.exception())
+
+
+def spawn_handle(name: str, fn: Callable) -> TaskHandle:
+    """Starts ``fn()`` as a background task and returns the task handle.
+    Unhandled exceptions are logged."""
+    task = asyncio.ensure_future(fn())
+    try:
+        task.set_name(name)
+    except AttributeError:
+        pass
+    task.add_done_callback(_log_task_exception)
+    return task
+
+
+async def join_handle(handle: TaskHandle, timeout: float) -> None:
+    """Waits up to ``timeout`` seconds for a spawned task to finish, mirroring
+    ``Thread.join(timeout)``: the task's exception (if any) is not re-raised.
+    On timeout the task is cancelled so it does not leak."""
+    try:
+        await asyncio.wait_for(asyncio.shield(handle), timeout)
+    except asyncio.TimeoutError:
+        handle.cancel()
+    except asyncio.CancelledError:
+        # The awaited task was cancelled elsewhere; do not treat that as a
+        # cancellation of the caller.
+        if handle.cancelled():
+            return
+        raise
+    except Exception:
+        pass
+
+
+class AsyncCallbackScheduler:
+    """Bridges sync notification paths to async callbacks: ``call`` schedules
+    a coroutine callback onto the event loop captured at construction time,
+    logging any unhandled exception. Safe to invoke from any thread."""
+
+    def __init__(self):
+        self._loop = asyncio.get_running_loop()
+
+    def call(self, fn: Callable, *args) -> None:
+        future = asyncio.run_coroutine_threadsafe(fn(*args), self._loop)
+        future.add_done_callback(self._log_exception)
+
+    @staticmethod
+    def _log_exception(future) -> None:
+        if not future.cancelled() and future.exception() is not None:
+            log.error("Unhandled exception in scheduled callback", exc_info=future.exception())
+
+
+class AsyncTaskRunner:
+    """Spawns named background tasks and stops them all on demand."""
+
+    def __init__(self):
+        self._tasks: Set[asyncio.Task] = set()
+        self._stopped = False
+
+    def spawn(self, name: str, fn: Callable) -> asyncio.Task:
+        """Starts ``fn()`` as a background task and returns the task handle.
+        Unhandled exceptions are logged."""
+        task = asyncio.ensure_future(fn())
+        try:
+            task.set_name(name)
+        except AttributeError:
+            pass
+        task.add_done_callback(self._on_done)
+        self._tasks.add(task)
+        return task
+
+    def _on_done(self, task: asyncio.Task) -> None:
+        self._tasks.discard(task)
+        if not task.cancelled() and task.exception() is not None:
+            log.error("Unhandled exception in background task", exc_info=task.exception())
+
+    def is_stopped(self) -> bool:
+        return self._stopped
+
+    async def stop_all(self, timeout: float = 1) -> None:
+        """Cancels all running tasks and waits for them to finish, logging a
+        warning for any that do not terminate within ``timeout`` seconds."""
+        self._stopped = True
+        tasks = list(self._tasks)
+        if not tasks:
+            return
+        for task in tasks:
+            task.cancel()
+        _, pending = await asyncio.wait(tasks, timeout=timeout)
+        for task in pending:
+            log.warning("Task %s did not terminate in time", task.get_name())
+
+
+class AsyncRepeatingTask:
+    """Calls a callback repeatedly at fixed intervals on a background task.
+    Mirrors the semantics of ``ldclient.impl.repeating_task.RepeatingTask``:
+    the interval is measured from the start of each invocation, exceptions
+    from the callback are logged, and ``stop()`` prevents any further
+    invocations but cannot be undone."""
+
+    def __init__(self, label: str, interval: float, initial_delay: float, callable: Callable):
+        self.__label = label
+        self.__interval = interval
+        self.__initial_delay = initial_delay
+        self.__action = callable
+        self.__stop = AsyncEvent()
+        self.__task: Optional[asyncio.Task] = None
+
+    def start(self):
+        """Starts the background task. Like a thread, the task can only be
+        started once."""
+        if self.__task is not None:
+            raise RuntimeError("tasks can only be started once")
+        self.__task = asyncio.ensure_future(self._run())
+        try:
+            self.__task.set_name(f"{self.__label}.repeating")
+        except AttributeError:
+            pass
+
+    def stop(self):
+        """Tells the background task to stop. It cannot be restarted after this."""
+        self.__stop.set()
+        task = self.__task
+        # When stop() is called from within the action itself, let the loop
+        # exit via the stop event rather than cancelling the current task.
+        if task is not None and task is not asyncio.current_task():
+            task.cancel()
+
+    async def _run(self):
+        try:
+            if self.__initial_delay > 0:
+                if await self.__stop.wait(self.__initial_delay):
+                    return
+            stopped = self.__stop.is_set()
+            while not stopped:
+                next_time = time.time() + self.__interval
+                try:
+                    result = self.__action()
+                    if inspect.isawaitable(result):
+                        await result
+                except asyncio.CancelledError:
+                    raise
+                except Exception as e:
+                    log.exception("Unexpected exception on worker task: %s" % e)
+                delay = next_time - time.time()
+                if delay > 0:
+                    stopped = await self.__stop.wait(delay)
+                else:
+                    # Yield to the event loop between back-to-back invocations
+                    await asyncio.sleep(0)
+                    stopped = self.__stop.is_set()
+        except asyncio.CancelledError:
+            pass
+
+
+class AsyncWorkerPool:
+    """A fixed-size pool of concurrent tasks that rejects jobs when its limit
+    is reached. Matches the contract of
+    ``ldclient.impl.fixed_thread_pool.FixedThreadPool``."""
+
+    def __init__(self, size: int, name: str):
+        self._size = size
+        self._name = name
+        self._busy: Set[asyncio.Task] = set()
+        self._event = AsyncEvent()
+        self._stopped = False
+
+    def execute(self, jobFn: Callable) -> bool:
+        """Schedules a job for execution if the pool is not already at its
+        limit, and returns True if successful; returns False if all workers
+        are busy."""
+        if self._stopped or len(self._busy) >= self._size:
+            return False
+        task = asyncio.ensure_future(self._run_job(jobFn))
+        self._busy.add(task)
+        return True
+
+    async def _run_job(self, jobFn: Callable) -> None:
+        try:
+            result = jobFn()
+            if inspect.isawaitable(result):
+                await result
+        except Exception:
+            log.warning('Unhandled exception in worker thread', exc_info=True)
+        finally:
+            task = asyncio.current_task()
+            if task is not None:
+                self._busy.discard(task)
+            self._event.set()
+
+    async def wait(self) -> None:
+        """Waits until all currently busy workers have completed their jobs."""
+        while len(self._busy) > 0:
+            self._event.clear()
+            if len(self._busy) == 0:
+                return
+            await self._event.wait()
+
+    def stop(self) -> None:
+        """Tells the pool to reject any further jobs; active jobs run to
+        completion."""
+        self._stopped = True
diff --git a/ldclient/impl/aio/transport.py b/ldclient/impl/aio/transport.py
new file mode 100644
index 00000000..1113464f
--- /dev/null
+++ b/ldclient/impl/aio/transport.py
@@ -0,0 +1,154 @@
+"""
+Async HTTP transport wrapping an aiohttp ``ClientSession``, exposing a
+``TransportResponse`` so callers can inspect a response after the request
+context has closed. The sync side talks to urllib3 directly.
+"""
+
+import ssl
+from typing import Optional, Union
+
+import aiohttp
+import certifi
+from ld_eventsource.async_client import AsyncSSEClient
+from ld_eventsource.config.async_connect_strategy import AsyncConnectStrategy
+from ld_eventsource.config.error_strategy import ErrorStrategy
+from ld_eventsource.config.retry_delay_strategy import RetryDelayStrategy
+
+from ldclient.impl.aio.transport_types import TransportResponse
+from ldclient.impl.http import _base_headers
+from ldclient.impl.util import log
+
+# Allows up to 5 minutes to elapse without any data sent across the stream.
+# Heartbeats sent as comments will keep this from triggering.
+STREAM_READ_TIMEOUT = 5 * 60
+
+MAX_RETRY_DELAY = 30
+BACKOFF_RESET_INTERVAL = 60
+JITTER_RATIO = 0.5
+
+
+def make_client_session(config, http_options=None) -> aiohttp.ClientSession:
+    """Creates an ``aiohttp.ClientSession`` configured from the SDK config's
+    HTTP options (CA certs, client certs, SSL verification, proxy trust).
+    ``http_options`` overrides the config's HTTP options when given."""
+    http_config = http_options if http_options is not None else config.http
+    ssl_ctx = ssl.create_default_context(cafile=http_config.ca_certs or certifi.where())
+    if http_config.cert_file:
+        ssl_ctx.load_cert_chain(http_config.cert_file)
+    if http_config.disable_ssl_verification:
+        ssl_ctx.check_hostname = False
+        ssl_ctx.verify_mode = ssl.CERT_NONE
+        log.warning("TLS verification disabled")
+
+    connector = aiohttp.TCPConnector(ssl=ssl_ctx, limit_per_host=10)
+    return aiohttp.ClientSession(
+        connector=connector,
+        trust_env=(http_config.http_proxy is None),
+    )
+
+
+class AsyncHTTPTransport:
+    """Performs HTTP requests over an aiohttp ``ClientSession``.
+
+    If no session is supplied, one is created lazily from the config on first
+    use and owned (closed) by the transport; a supplied session remains owned
+    by the caller. ``http_options`` overrides the config's HTTP options for
+    lazy session construction and default request timeouts.
+    """
+
+    def __init__(
+        self,
+        config,
+        client: Optional[aiohttp.ClientSession] = None,
+        http_options=None,
+    ):
+        self._config = config
+        self._http_options = http_options if http_options is not None else config.http
+        self._client = client
+        self._owns_client = client is None
+        self._proxy = self._http_options.http_proxy or None
+
+    async def request(
+        self,
+        method: str,
+        uri: str,
+        headers: Optional[dict] = None,
+        body: Optional[Union[bytes, str]] = None,
+        connect_timeout: Optional[float] = None,
+        read_timeout: Optional[float] = None,
+    ) -> TransportResponse:
+        """Performs a request."""
+        if self._client is None:
+            self._client = make_client_session(self._config, self._http_options)
+        timeout = aiohttp.ClientTimeout(
+            connect=connect_timeout if connect_timeout is not None else self._http_options.connect_timeout,
+            sock_read=read_timeout if read_timeout is not None else self._http_options.read_timeout,
+        )
+        async with self._client.request(
+            method,
+            uri,
+            headers=headers,
+            data=body,
+            timeout=timeout,
+            proxy=self._proxy,
+        ) as response:
+            text = await response.text(encoding='UTF-8', errors='replace')
+            return TransportResponse(response.status, response.headers, text)
+
+    async def close(self) -> None:
+        """Closes the underlying session if this transport created it."""
+        if self._owns_client and self._client is not None:
+            await self._client.close()
+            self._client = None
+
+
+class AsyncSSEFactory:
+    """Creates configured ``AsyncSSEClient`` instances for streaming connections.
+
+    A supplied aiohttp session remains owned by the caller; the SSE client
+    never closes it. Callers are expected to supply a session built from the
+    SDK's HTTP options (via ``make_client_session``) so the streaming
+    connection uses the configured certs, SSL settings, and proxy trust.
+    ``http_options`` overrides the config's HTTP options for connection
+    timeouts and proxy settings.
+    """
+
+    def __init__(self, config, session: Optional[aiohttp.ClientSession] = None, proxy: Optional[str] = None, http_options=None):
+        self._config = config
+        self._session = session
+        self._http_options = http_options if http_options is not None else config.http
+        self._proxy = proxy if proxy is not None else (self._http_options.http_proxy or None)
+
+    def create(self, url: str, initial_retry_delay: float, query_params=None) -> AsyncSSEClient:
+        """Builds an SSE client for the given stream URL. Headers, timeouts,
+        proxy settings, and the retry/backoff policy come from the SDK config.
+        ``query_params`` is an optional zero-argument callable evaluated on
+        each (re)connect to produce additional query string parameters."""
+        base_headers = _base_headers(self._config)
+        aiohttp_request_options: dict = {
+            "timeout": aiohttp.ClientTimeout(
+                total=None,
+                connect=self._http_options.connect_timeout,
+                sock_read=STREAM_READ_TIMEOUT,
+            )
+        }
+        if self._proxy:
+            aiohttp_request_options["proxy"] = self._proxy
+        return AsyncSSEClient(
+            connect=AsyncConnectStrategy.http(
+                url=url,
+                headers=base_headers,
+                session=self._session,
+                aiohttp_request_options=aiohttp_request_options,
+                query_params=query_params,
+            ),
+            error_strategy=ErrorStrategy.always_continue(),  # we'll make error-handling decisions when we see a Fault
+            initial_retry_delay=initial_retry_delay,
+            retry_delay_strategy=RetryDelayStrategy.default(
+                max_delay=MAX_RETRY_DELAY,
+                backoff_multiplier=2,
+                jitter_multiplier=JITTER_RATIO,
+            ),
+            retry_delay_reset_threshold=BACKOFF_RESET_INTERVAL,
+            logger=log,
+        )
diff --git a/ldclient/impl/aio/transport_types.py b/ldclient/impl/aio/transport_types.py
new file mode 100644
index 00000000..e39d243c
--- /dev/null
+++ b/ldclient/impl/aio/transport_types.py
@@ -0,0 +1,19 @@
+"""
+Shared types for the async/sync transport shims.
+"""
+
+from typing import Mapping
+
+
+class TransportResponse:
+    """A minimal uniform HTTP response: status code, headers, and decoded body.
+
+    ``headers`` is whatever case-insensitive mapping the underlying HTTP
+    library produced, so lookups like ``headers.get('ETag')`` work regardless
+    of the casing the server used.
+    """
+
+    def __init__(self, status: int, headers: Mapping[str, str], body: str):
+        self.status = status
+        self.headers = headers
+        self.body = body
diff --git a/ldclient/testing/test_aio.py b/ldclient/testing/test_aio.py
new file mode 100644
index 00000000..65ecdfe1
--- /dev/null
+++ b/ldclient/testing/test_aio.py
@@ -0,0 +1,428 @@
+"""
+Tests for the async concurrency primitives (``aio.concurrency``) and the async
+transport (``aio.transport``) in ldclient/impl/aio/. The sync side uses
+stdlib/SDK primitives directly, so there is no sync twin to assert parity
+against.
+"""
+
+import asyncio
+import threading
+import time
+
+import pytest
+
+from ldclient.config import Config
+from ldclient.impl.aio import concurrency as aio
+from ldclient.impl.aio.transport import AsyncHTTPTransport
+from ldclient.testing.http_util import (
+    BasicResponse,
+    JsonResponse,
+    start_server
+)
+
+
+async def _async_wait_until(predicate, timeout=2.0):
+    deadline = time.time() + timeout
+    while not predicate():
+        assert time.time() < deadline, "timed out waiting for condition"
+        await asyncio.sleep(0.01)
+
+
+# ---------------------------------------------------------------------------
+# Event
+# ---------------------------------------------------------------------------
+
+class TestEventParity:
+    @pytest.mark.asyncio
+    async def test_async_set_clear_is_set(self):
+        event = aio.AsyncEvent()
+        assert not event.is_set()
+        event.set()
+        assert event.is_set()
+        assert await event.wait(1) is True
+        event.clear()
+        assert not event.is_set()
+
+    @pytest.mark.asyncio
+    async def test_async_wait_timeout_returns_false(self):
+        event = aio.AsyncEvent()
+        start = time.time()
+        assert await event.wait(0.05) is False
+        assert time.time() - start >= 0.04
+
+    @pytest.mark.asyncio
+    async def test_async_wait_wakes_when_set(self):
+        event = aio.AsyncEvent()
+
+        async def setter():
+            await asyncio.sleep(0.02)
+            event.set()
+
+        task = asyncio.ensure_future(setter())
+        assert await event.wait(2) is True
+        await task
+
+
+# ---------------------------------------------------------------------------
+# Lock
+# ---------------------------------------------------------------------------
+
+class TestLockParity:
+    @pytest.mark.asyncio
+    async def test_async_lock_context_manager(self):
+        lock = aio.AsyncLock()
+        assert not lock.locked()
+        async with lock:
+            assert lock.locked()
+        assert not lock.locked()
+
+    @pytest.mark.asyncio
+    async def test_async_lock_mutual_exclusion(self):
+        lock = aio.AsyncLock()
+        counter = {'value': 0, 'concurrent': 0, 'max_concurrent': 0}
+
+        async def work():
+            async with lock:
+                counter['concurrent'] += 1
+                counter['max_concurrent'] = max(counter['max_concurrent'], counter['concurrent'])
+                await asyncio.sleep(0.01)
+                counter['value'] += 1
+                counter['concurrent'] -= 1
+
+        await asyncio.gather(*(work() for _ in range(5)))
+        assert counter['value'] == 5
+        assert counter['max_concurrent'] == 1
+
+
+# ---------------------------------------------------------------------------
+# Queue
+# ---------------------------------------------------------------------------
+
+class TestQueueParity:
+    @pytest.mark.asyncio
+    async def test_async_put_get_and_empty(self):
+        q = aio.AsyncQueue()
+        assert q.empty()
+        await q.put('a')
+        assert not q.empty()
+        assert await q.get(1) == 'a'
+        assert q.empty()
+
+    @pytest.mark.asyncio
+    async def test_async_get_timeout_raises_queue_empty(self):
+        q = aio.AsyncQueue()
+        with pytest.raises(aio.QueueEmpty):
+            await q.get(0.05)
+
+    @pytest.mark.asyncio
+    async def test_async_get_nonblocking_raises_queue_empty(self):
+        q = aio.AsyncQueue()
+        with pytest.raises(aio.QueueEmpty):
+            await q.get(block=False)
+
+    @pytest.mark.asyncio
+    async def test_async_put_nowait_raises_queue_full_at_capacity(self):
+        q = aio.AsyncQueue(1)
+        q.put_nowait('a')
+        with pytest.raises(aio.QueueFull):
+            q.put_nowait('b')
+        assert await q.get(block=False) == 'a'
+
+
+# ---------------------------------------------------------------------------
+# RepeatingTask
+# ---------------------------------------------------------------------------
+
+class TestRepeatingTaskParity:
+    @pytest.mark.asyncio
+    async def test_async_fires_repeatedly_then_stops(self):
+        counts = {'n': 0}
+
+        async def action():
+            counts['n'] += 1
+
+        task = aio.AsyncRepeatingTask("test.repeating", 0.01, 0, action)
+        task.start()
+        await _async_wait_until(lambda: counts['n'] >= 3)
+        task.stop()
+        await asyncio.sleep(0.05)
+        snapshot = counts['n']
+        await asyncio.sleep(0.05)
+        assert counts['n'] == snapshot
+
+    @pytest.mark.asyncio
+    async def test_async_initial_delay_respected(self):
+        counts = {'n': 0}
+
+        async def action():
+            counts['n'] += 1
+
+        task = aio.AsyncRepeatingTask("test.repeating", 0.01, 0.1, action)
+        task.start()
+        await asyncio.sleep(0.03)
+        assert counts['n'] == 0
+        task.stop()
+
+    @pytest.mark.asyncio
+    async def test_async_continues_after_action_exception(self):
+        counts = {'n': 0}
+
+        async def action():
+            counts['n'] += 1
+            raise RuntimeError("boom")
+
+        task = aio.AsyncRepeatingTask("test.repeating", 0.01, 0, action)
+        task.start()
+        await _async_wait_until(lambda: counts['n'] >= 2)
+        task.stop()
+
+    @pytest.mark.asyncio
+    async def test_async_stop_from_within_action(self):
+        counts = {'n': 0}
+        holder = {}
+
+        async def action():
+            counts['n'] += 1
+            holder['task'].stop()
+
+        holder['task'] = aio.AsyncRepeatingTask("test.repeating", 0.01, 0, action)
+        holder['task'].start()
+        await asyncio.sleep(0.1)
+        assert counts['n'] == 1
+
+    @pytest.mark.asyncio
+    async def test_async_second_start_raises(self):
+        async def action():
+            pass
+
+        task = aio.AsyncRepeatingTask("test.repeating", 0.01, 0, action)
+        task.start()
+        with pytest.raises(RuntimeError):
+            task.start()
+        task.stop()
+
+
+# ---------------------------------------------------------------------------
+# WorkerPool
+# ---------------------------------------------------------------------------
+
+class TestWorkerPoolParity:
+    @pytest.mark.asyncio
+    async def test_async_saturation_returns_false(self):
+        pool = aio.AsyncWorkerPool(1, "test.pool")
+        release = aio.AsyncEvent()
+        started = aio.AsyncEvent()
+
+        async def job():
+            started.set()
+            await release.wait(2)
+
+        assert pool.execute(job) is True
+        await started.wait(2)
+        assert pool.execute(lambda: None) is False
+        release.set()
+        await pool.wait()
+
+        async def noop():
+            pass
+
+        assert pool.execute(noop) is True
+        await pool.wait()
+        pool.stop()
+
+    @pytest.mark.asyncio
+    async def test_async_wait_returns_when_idle(self):
+        pool = aio.AsyncWorkerPool(2, "test.pool")
+        await pool.wait()
+        pool.stop()
+
+
+# ---------------------------------------------------------------------------
+# TaskRunner
+# ---------------------------------------------------------------------------
+
+class TestTaskRunnerParity:
+    @pytest.mark.asyncio
+    async def test_async_spawn_runs_function(self):
+        runner = aio.AsyncTaskRunner()
+        done = aio.AsyncEvent()
+
+        async def fn():
+            done.set()
+
+        runner.spawn("test.task", fn)
+        assert await done.wait(2)
+        await runner.stop_all()
+        assert runner.is_stopped()
+
+    @pytest.mark.asyncio
+    async def test_async_stop_all_cancels_running_tasks(self):
+        runner = aio.AsyncTaskRunner()
+        started = aio.AsyncEvent()
+
+        async def forever():
+            started.set()
+            await asyncio.sleep(60)
+
+        task = runner.spawn("test.forever", forever)
+        await started.wait(2)
+        await runner.stop_all()
+        assert task.done()
+
+    @pytest.mark.asyncio
+    async def test_async_stop_all_accepts_timeout(self):
+        runner = aio.AsyncTaskRunner()
+        started = aio.AsyncEvent()
+        give_up = asyncio.Event()
+
+        async def stubborn():
+            while not give_up.is_set():
+                try:
+                    started.set()
+                    await asyncio.sleep(60)
+                except asyncio.CancelledError:
+                    continue  # refuse to die until give_up is set
+
+        task = runner.spawn("test.stubborn", stubborn)
+        await started.wait(2)
+        start = time.time()
+        await runner.stop_all(timeout=0.1)
+        assert time.time() - start < 2
+        assert not task.done()  # the stubborn task outlived stop_all
+        give_up.set()
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+
+# ---------------------------------------------------------------------------
+# spawn_handle / join_handle
+# ---------------------------------------------------------------------------
+
+class TestHandleParity:
+    @pytest.mark.asyncio
+    async def test_async_spawn_and_join(self):
+        done = aio.AsyncEvent()
+
+        async def fn():
+            done.set()
+
+        handle = aio.spawn_handle("test.handle", fn)
+        await aio.join_handle(handle, 2)
+        assert done.is_set()
+        assert handle.done()
+
+    @pytest.mark.asyncio
+    async def test_async_join_times_out_and_cancels(self):
+        started = aio.AsyncEvent()
+
+        async def slow():
+            started.set()
+            await asyncio.sleep(60)
+
+        handle = aio.spawn_handle("test.slow", slow)
+        await started.wait(2)
+        await aio.join_handle(handle, 0.1)
+        await _async_wait_until(handle.done)
+        assert handle.cancelled()
+
+    @pytest.mark.asyncio
+    async def test_async_join_swallows_task_exception(self):
+        async def boom():
+            raise ValueError("boom")
+
+        handle = aio.spawn_handle("test.boom", boom)
+        await aio.join_handle(handle, 2)  # does not raise
+
+
+# ---------------------------------------------------------------------------
+# Callback scheduler
+# ---------------------------------------------------------------------------
+
+class TestCallbackSchedulerParity:
+    @pytest.mark.asyncio
+    async def test_async_call_schedules_coroutine_with_args(self):
+        scheduler = aio.AsyncCallbackScheduler()
+        received = []
+
+        async def cb(value):
+            received.append(value)
+
+        scheduler.call(cb, 'value')
+        await _async_wait_until(lambda: received == ['value'])
+
+    @pytest.mark.asyncio
+    async def test_async_call_works_from_worker_thread(self):
+        scheduler = aio.AsyncCallbackScheduler()
+        received = []
+
+        async def cb(value):
+            received.append(value)
+
+        thread = threading.Thread(target=lambda: scheduler.call(cb, 'threaded'))
+        thread.start()
+        thread.join()
+        await _async_wait_until(lambda: received == ['threaded'])
+
+    @pytest.mark.asyncio
+    async def test_async_call_logs_callback_exception(self):
+        scheduler = aio.AsyncCallbackScheduler()
+        completed = aio.AsyncEvent()
+
+        async def boom():
+            completed.set()
+            raise ValueError("boom")
+
+        scheduler.call(boom)  # must not raise
+        assert await completed.wait(2)
+        await asyncio.sleep(0.05)  # let the done callback run
+
+
+# ---------------------------------------------------------------------------
+# HTTP transport
+# ---------------------------------------------------------------------------
+
+class TestTransportParity:
+    @pytest.mark.asyncio
+    async def test_async_get(self):
+        with start_server() as server:
+            config = Config(sdk_key='sdk-key', base_uri=server.uri)
+            transport = AsyncHTTPTransport(config)
+            server.for_path('/path', JsonResponse({'hello': 'world'}, {'Etag': 'my-etag'}))
+
+            resp = await transport.request('GET', server.uri + '/path', headers={'X-Test': 'yes'})
+
+            assert resp.status == 200
+            assert resp.headers.get('ETag') == 'my-etag'
+            assert '"hello"' in resp.body
+            req = server.require_request()
+            assert req.headers['X-Test'] == 'yes'
+            await transport.close()
+
+    @pytest.mark.asyncio
+    async def test_async_post_with_body(self):
+        with start_server() as server:
+            config = Config(sdk_key='sdk-key', base_uri=server.uri)
+            transport = AsyncHTTPTransport(config)
+            server.for_path('/events', BasicResponse(202))
+
+            resp = await transport.request('POST', server.uri + '/events', headers={'Content-Type': 'application/json'}, body='[{"kind":"custom"}]')
+
+            assert resp.status == 202
+            req = server.require_request()
+            assert req.method == 'POST'
+            assert req.body == '[{"kind":"custom"}]'
+            await transport.close()
+
+    @pytest.mark.asyncio
+    async def test_async_error_status_is_returned_not_raised(self):
+        with start_server() as server:
+            config = Config(sdk_key='sdk-key', base_uri=server.uri)
+            transport = AsyncHTTPTransport(config)
+            server.for_path('/missing', BasicResponse(404))
+
+            resp = await transport.request('GET', server.uri + '/missing')
+            assert resp.status == 404
+            await transport.close()
diff --git a/pyproject.toml b/pyproject.toml
index 8672bb94..46b629bc 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -36,6 +36,7 @@ Repository = "https://github.com/launchdarkly/python-server-sdk"
 Documentation = "https://launchdarkly-python-sdk.readthedocs.io/en/latest/"
 
 [project.optional-dependencies]
+async = ["aiohttp>=3.9,<4"]
 redis = ["redis>=2.10.5"]
 consul = ["python-consul>=1.0.1"]
 dynamodb = ["boto3>=1.9.71"]
@@ -45,6 +46,7 @@ test-filesource = ["pyyaml>=5.3.1", "watchdog>=3.0.0"]
 dev = [
     "mock>=2.0.0",
     "pytest>=8.0.0,<10",
+    "pytest-asyncio>=0.23",
     "redis>=2.10.5,<9.0.0",
     "boto3>=1.9.71,<2.0.0",
     "coverage>=4.4",
@@ -58,6 +60,7 @@ dev = [
     "types-mock>=5.0",
     "types-redis>=4.0",
     "types-setuptools>=68.0",
+    "aiohttp>=3.8.0",
 ]
 contract-tests = [
     "Flask<4",
@@ -85,6 +88,7 @@ multi_line_output = 3
 
 [tool.pytest.ini_options]
 addopts = ["-ra"]
+asyncio_mode = "strict"
 
 
 [build-system]