diff --git a/pyproject.toml b/pyproject.toml
index 8af3453..0737ecd 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -48,6 +48,7 @@ config = ["tomli>=2.0.0; python_version < '3.11'", "tomli-w>=1.0.0"]
 dev = [
     "pytest>=7.0",
     "pytest-cov>=4.0",
+    "pytest-asyncio>=0.23",
     "ruff>=0.1.0",
     "tomli-w>=1.0.0",
     # the backends exercised by the parametrized test suite
@@ -94,6 +95,9 @@ convention = "google"
 minversion = "6.0"
 testpaths = ["tests"]
 doctest_optionflags = ["NORMALIZE_WHITESPACE", "ELLIPSIS"]
+# Make every `async def test_*` run automatically — no per-test decorator
+# needed. Used by tests/test_async.py.
+asyncio_mode = "auto"
 
 [tool.wads.ci]
 installer = "uv"
diff --git a/tests/test_async.py b/tests/test_async.py
new file mode 100644
index 0000000..5db4f5c
--- /dev/null
+++ b/tests/test_async.py
@@ -0,0 +1,234 @@
+"""
+Contract tests for the async wrapper, ``connect_async``, and ``hybrid_search_async``.
+
+Phase 1 of #18 — the universal :class:`~vd.asynchronous.AsyncClientWrapper`
+adapts every sync backend to the :class:`vd.AsyncClient` /
+:class:`vd.AsyncCollection` contract via :func:`asyncio.to_thread`. These
+tests run against the ``memory`` backend (no infra needed) and verify:
+
+- the entry point :func:`vd.connect_async` returns an awaitable that resolves
+  to an :class:`~vd.AsyncClient`;
+- the wrapper satisfies :class:`vd.AsyncClient` / :class:`vd.AsyncCollection`
+  structurally;
+- ``native_async`` is ``False`` for the wrapper (signals to_thread-based);
+- get / set / delete / keys / count / search all work and are awaitable /
+  async-iterable;
+- :func:`vd.hybrid_search_async` dispatches correctly through the wrapper;
+- the async context-manager protocol works.
+
+Phase 2 follow-ups will add per-backend native async adapters; each backend
+will get its own parametrized test entries at that time.
+"""
+
+import pytest
+
+import vd
+
+pytestmark = pytest.mark.asyncio
+
+
+# --------------------------------------------------------------------------- #
+# Entry point + protocol satisfaction
+# --------------------------------------------------------------------------- #
+
+
+async def test_connect_async_returns_async_client():
+    client = await vd.connect_async("memory")
+    try:
+        assert isinstance(client, vd.AsyncClient)
+        # The universal wrapper structurally satisfies SupportsNativeAsync
+        # (the attribute exists); native_async is False to signal that I/O
+        # runs in a thread pool.
+        assert isinstance(client, vd.SupportsNativeAsync)
+        assert client.native_async is False
+    finally:
+        await client.close()
+
+
+async def test_async_context_manager():
+    async with await vd.connect_async("memory") as client:
+        assert isinstance(client, vd.AsyncClient)
+        col = await client.create_collection("ctx", dimension=2)
+        assert isinstance(col, vd.AsyncCollection)
+
+
+# --------------------------------------------------------------------------- #
+# AsyncClient surface
+# --------------------------------------------------------------------------- #
+
+
+async def test_create_get_delete_list_collections():
+    async with await vd.connect_async("memory") as client:
+        # Initially empty
+        names = [n async for n in client.list_collections()]
+        assert names == []
+
+        col = await client.create_collection("docs", dimension=2)
+        assert isinstance(col, vd.AsyncCollection)
+
+        names = [n async for n in client.list_collections()]
+        assert "docs" in names
+
+        # Round-trip via get_collection
+        same = await client.get_collection("docs")
+        assert isinstance(same, vd.AsyncCollection)
+
+        # get_or_create returns existing
+        again = await client.get_or_create_collection("docs", dimension=2)
+        assert isinstance(again, vd.AsyncCollection)
+
+        # Drop
+        await client.delete_collection("docs")
+        names = [n async for n in client.list_collections()]
+        assert "docs" not in names
+
+
+async def test_get_missing_collection_raises_keyerror():
+    async with await vd.connect_async("memory") as client:
+        with pytest.raises(KeyError):
+            await client.get_collection("nope")
+
+
+# --------------------------------------------------------------------------- #
+# AsyncCollection surface
+# --------------------------------------------------------------------------- #
+
+
+async def _populate(col):
+    """Write 3 docs into ``col``."""
+    await col.set("a", vd.Document(id="a", text="cats purr", vector=[1.0, 0.0]))
+    await col.set("b", vd.Document(id="b", text="dogs bark", vector=[0.0, 1.0]))
+    await col.set("c", vd.Document(id="c", text="cats and dogs", vector=[0.5, 0.5]))
+
+
+async def test_set_get_count_keys_delete():
+    async with await vd.connect_async("memory") as client:
+        col = await client.create_collection("crud", dimension=2)
+        await _populate(col)
+
+        assert await col.count() == 3
+
+        keys = sorted([k async for k in col.keys()])
+        assert keys == ["a", "b", "c"]
+
+        doc = await col.get("a")
+        assert isinstance(doc, vd.Document)
+        assert doc.id == "a"
+        assert doc.text == "cats purr"
+
+        await col.delete("b")
+        assert await col.count() == 2
+        with pytest.raises(KeyError):
+            await col.get("b")
+
+
+async def test_set_accepts_text_and_tuple_inputs():
+    """The wrapper passes flexible inputs straight to sync __setitem__."""
+    async with await vd.connect_async("memory") as client:
+        col = await client.create_collection(
+            "flex", dimension=2, metric="cosine"
+        )
+        # Pre-vector required for memory (no embedder); use Document form.
+        await col.set("a", vd.Document(id="a", text="hi", vector=[1.0, 0.0]))
+        doc = await col.get("a")
+        assert doc.text == "hi"
+
+
+async def test_search_returns_async_iterator():
+    async with await vd.connect_async("memory") as client:
+        col = await client.create_collection("srch", dimension=2)
+        await _populate(col)
+        hits = []
+        async for h in col.search([0.9, 0.1], limit=2):
+            hits.append(h)
+        assert len(hits) == 2
+        # Closest to [0.9, 0.1] is 'a' ([1, 0]); next is 'c' ([0.5, 0.5]).
+        assert hits[0]["id"] == "a"
+        # Descending score order
+        assert hits[0]["score"] >= hits[1]["score"]
+
+
+async def test_search_filter_and_egress():
+    async with await vd.connect_async("memory") as client:
+        col = await client.create_collection("filt", dimension=2)
+        await col.set("a", vd.Document(
+            id="a", text="x", vector=[1.0, 0.0], metadata={"k": 1}
+        ))
+        await col.set("b", vd.Document(
+            id="b", text="y", vector=[0.0, 1.0], metadata={"k": 2}
+        ))
+        # Filter to k==2 only
+        ids = [h["id"] async for h in col.search([0.9, 0.1], filter={"k": 2})]
+        assert ids == ["b"]
+        # Egress transform
+        only_ids = [
+            r async for r in col.search([0.9, 0.1], egress=lambda h: h["id"])
+        ]
+        assert set(only_ids) == {"a", "b"}
+
+
+async def test_upsert_and_add_documents_batch():
+    async with await vd.connect_async("memory") as client:
+        col = await client.create_collection("batch", dimension=2)
+        # upsert
+        await col.upsert(
+            vd.Document(id="x", text="x", vector=[1.0, 0.0])
+        )
+        # add_documents — Documents with vectors, no embedder needed
+        await col.add_documents(
+            [
+                vd.Document(id="y", text="y", vector=[0.0, 1.0]),
+                vd.Document(id="z", text="z", vector=[0.5, 0.5]),
+            ],
+            batch_size=2,
+        )
+        assert await col.count() == 3
+
+
+# --------------------------------------------------------------------------- #
+# Escape hatches
+# --------------------------------------------------------------------------- #
+
+
+async def test_sync_and_native_escape_hatches():
+    async with await vd.connect_async("memory") as client:
+        # AsyncClientWrapper.sync exposes the underlying sync Client
+        assert isinstance(client.sync, vd.Client)
+        # client.client mirrors the sync client's `client` property
+        # (None for memory; the attribute existing is the contract).
+        assert hasattr(client, "client")
+        col = await client.create_collection("esc", dimension=2)
+        # AsyncCollectionWrapper.sync exposes the underlying sync Collection
+        assert isinstance(col.sync, vd.Collection)
+        assert hasattr(col, "native")
+
+
+# --------------------------------------------------------------------------- #
+# hybrid_search_async
+# --------------------------------------------------------------------------- #
+
+
+async def test_hybrid_search_async_basic():
+    async with await vd.connect_async("memory") as client:
+        col = await client.create_collection("hyb", dimension=2)
+        await _populate(col)
+        hits = []
+        async for h in vd.hybrid_search_async(
+            col, [0.9, 0.1], query_text="cats", limit=3
+        ):
+            hits.append(h)
+        ids = [h["id"] for h in hits]
+        # 'a' has both dense + lexical signal — should top the fused ranking.
+        assert "a" in ids[:2]
+        # Each result is the standard contract shape
+        for h in hits:
+            assert {"id", "text", "score", "metadata"} <= set(h.keys())
+
+
+async def test_hybrid_search_async_requires_query_text_for_vector_query():
+    async with await vd.connect_async("memory") as client:
+        col = await client.create_collection("hyb_err", dimension=2)
+        await _populate(col)
+        with pytest.raises(ValueError, match="query_text"):
+            async for _ in vd.hybrid_search_async(col, [0.9, 0.1], limit=2):
+                pass
diff --git a/vd/__init__.py b/vd/__init__.py
index 2a3a298..8ff185e 100644
--- a/vd/__init__.py
+++ b/vd/__init__.py
@@ -50,6 +50,8 @@ def skills_dir() -> _Path:
 from vd.base import (  # noqa: E402
     AbstractClient,
     AbstractCollection,
+    AsyncClient,
+    AsyncCollection,
     BackendNotInstalledError,
     Client,
     Collection,
@@ -62,12 +64,21 @@ def skills_dir() -> _Path:
     StaticIndexError,
     SupportsBatch,
     SupportsHybrid,
+    SupportsNativeAsync,
     UnsupportedCapabilityError,
     UnsupportedFilterError,
     VdError,
     Vector,
 )
 
+# ----- async support ------------------------------------------------------- #
+from vd.asynchronous import (  # noqa: E402
+    AsyncClientWrapper,
+    AsyncCollectionWrapper,
+    connect_async,
+    hybrid_search_async,
+)
+
 # ----- the entry point & registry ------------------------------------------ #
 from vd.util import (  # noqa: E402
     connect,
@@ -217,6 +228,14 @@ def skills_dir() -> _Path:
     # capability protocols
     "SupportsBatch",
     "SupportsHybrid",
+    "SupportsNativeAsync",
+    # async
+    "AsyncClient",
+    "AsyncCollection",
+    "AsyncClientWrapper",
+    "AsyncCollectionWrapper",
+    "connect_async",
+    "hybrid_search_async",
     # filter language
     "matches_filter",
     "validate_filter",
diff --git a/vd/asynchronous.py b/vd/asynchronous.py
new file mode 100644
index 0000000..7ea08d2
--- /dev/null
+++ b/vd/asynchronous.py
@@ -0,0 +1,405 @@
+"""
+Async support for ``vd``: universal wrapper + opt-in native implementations.
+
+This module gives every ``vd`` backend an ``async``/``await`` surface day one,
+without forking the adapter hierarchy. Two pieces:
+
+- :class:`AsyncCollectionWrapper` / :class:`AsyncClientWrapper` —
+  thin adapters that take any sync :class:`vd.Collection` / :class:`vd.Client`
+  and dispatch every method to :func:`asyncio.to_thread`. This is the
+  **universal fallback**: every backend works through it.
+- :func:`connect_async` — the entry point. Mirrors :func:`vd.connect`. If a
+  backend ships a native async client (Phase 2 follow-ups: chroma, qdrant,
+  weaviate, elasticsearch, redis, mongodb, lancedb, milvus, pinecone,
+  turbopuffer), :func:`connect_async` returns *that*; otherwise it returns
+  the wrapper.
+
+The asyncio.to_thread wrapper does **not** unblock the event loop — it just
+moves blocking calls off the main thread, freeing the loop. For real
+non-blocking I/O against a network backend, use a client that satisfies
+:class:`vd.SupportsNativeAsync`.
+
+The module name is ``vd.asynchronous`` (not ``vd.async``) because ``async``
+is a Python keyword.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from typing import Any, AsyncIterator, Callable, Iterable, Optional, Union
+
+from vd.base import (
+    AsyncClient,
+    AsyncCollection,
+    Document,
+    Filter,
+    SearchResult,
+    SupportsHybrid,
+    Vector,
+)
+
+# --------------------------------------------------------------------------- #
+# Universal wrappers
+# --------------------------------------------------------------------------- #
+
+
+class AsyncCollectionWrapper:
+    """
+    Adapt a sync :class:`~vd.Collection` to the :class:`~vd.AsyncCollection`
+    contract by dispatching every method to :func:`asyncio.to_thread`.
+
+    Use :func:`connect_async` rather than instantiating this directly — it
+    will pick this wrapper or a native async adapter as appropriate.
+
+    Parameters
+    ----------
+    sync_collection :
+        A live :class:`~vd.Collection` (typically obtained from a
+        :class:`~vd.Client`).
+
+    Attributes
+    ----------
+    native_async : bool
+        Always ``False`` for this wrapper. The wrapper still satisfies
+        :class:`~vd.SupportsNativeAsync` structurally (the attribute is
+        present), but the boolean tells callers that I/O is happening in a
+        thread pool rather than on the event loop. Prefer a native
+        implementation for high-concurrency workloads.
+    """
+
+    #: This wrapper offloads to a thread pool; it doesn't do non-blocking I/O.
+    native_async: bool = False
+
+    def __init__(self, sync_collection: Any):
+        self._sync = sync_collection
+
+    # ----- escape hatch — the wrapped sync collection ---------------------- #
+
+    @property
+    def sync(self) -> Any:
+        """The underlying sync :class:`~vd.Collection` — a documented escape hatch."""
+        return self._sync
+
+    @property
+    def native(self) -> Any:
+        """Pass through to the wrapped collection's :attr:`~vd.Collection.native`."""
+        return getattr(self._sync, "native", None)
+
+    # ----- AsyncCollection contract ---------------------------------------- #
+
+    async def get(self, key: str) -> Document:
+        """Fetch one document; raises ``KeyError`` if absent."""
+        return await asyncio.to_thread(self._sync.__getitem__, key)
+
+    async def set(
+        self, key: str, value: Union[str, tuple, Document]
+    ) -> None:
+        """Insert or replace a document (idempotent upsert)."""
+        await asyncio.to_thread(self._sync.__setitem__, key, value)
+
+    async def delete(self, key: str) -> None:
+        """Delete a document; raises ``KeyError`` if absent."""
+        await asyncio.to_thread(self._sync.__delitem__, key)
+
+    async def keys(self) -> AsyncIterator[str]:
+        """Yield document ids."""
+        # We materialize once in a worker thread, then yield from memory.
+        # Streaming through asyncio.to_thread per-item would be much slower
+        # for the common case where _keys() is already O(N) iteration.
+        ids = await asyncio.to_thread(lambda: list(self._sync))
+        for doc_id in ids:
+            yield doc_id
+
+    async def count(self) -> int:
+        """Return the number of documents."""
+        return await asyncio.to_thread(self._sync.__len__)
+
+    async def search(
+        self,
+        query: Union[str, Vector],
+        *,
+        limit: int = 10,
+        filter: Optional[Filter] = None,
+        egress: Optional[Callable[[SearchResult], Any]] = None,
+        **kwargs,
+    ) -> AsyncIterator[SearchResult]:
+        """
+        Yield the ``limit`` documents most similar to ``query``.
+
+        The underlying search runs once on a worker thread; results stream
+        from memory. (Most backends' sync ``search`` already returns a list
+        or a fully-realized iterator under the hood.)
+        """
+
+        def _run() -> list[SearchResult]:
+            return list(
+                self._sync.search(
+                    query, limit=limit, filter=filter, egress=egress, **kwargs
+                )
+            )
+
+        results = await asyncio.to_thread(_run)
+        for hit in results:
+            yield hit
+
+    # ----- batch convenience (also satisfies an async SupportsBatch) ------ #
+
+    async def add_documents(
+        self,
+        documents: Iterable[Any],
+        *,
+        batch_size: int = 100,
+    ) -> None:
+        """Batch upsert — mirrors :meth:`~vd.AbstractCollection.add_documents`."""
+        await asyncio.to_thread(
+            self._sync.add_documents, list(documents), batch_size=batch_size
+        )
+
+    async def upsert(self, document: Document) -> None:
+        """Insert or replace ``document``."""
+        await asyncio.to_thread(self._sync.upsert, document)
+
+
+class AsyncClientWrapper:
+    """
+    Adapt a sync :class:`~vd.Client` to the :class:`~vd.AsyncClient`
+    contract by dispatching every method to :func:`asyncio.to_thread`.
+
+    Use :func:`connect_async` rather than instantiating this directly.
+
+    Parameters
+    ----------
+    sync_client :
+        A live :class:`~vd.Client` (typically obtained from :func:`vd.connect`).
+
+    Attributes
+    ----------
+    native_async : bool
+        Always ``False`` for this wrapper.
+    """
+
+    native_async: bool = False
+
+    def __init__(self, sync_client: Any):
+        self._sync = sync_client
+
+    # ----- escape hatches -------------------------------------------------- #
+
+    @property
+    def sync(self) -> Any:
+        """The underlying sync :class:`~vd.Client` — a documented escape hatch."""
+        return self._sync
+
+    @property
+    def client(self) -> Any:
+        """Pass through to the wrapped client's :attr:`~vd.Client.client`."""
+        return getattr(self._sync, "client", None)
+
+    # ----- AsyncClient contract -------------------------------------------- #
+
+    async def create_collection(
+        self,
+        name: str,
+        *,
+        dimension: Optional[int] = None,
+        metric: str = "cosine",
+        **index_config,
+    ) -> AsyncCollection:
+        """Create a new collection; raise ``ValueError`` if it exists."""
+        col = await asyncio.to_thread(
+            self._sync.create_collection,
+            name,
+            dimension=dimension,
+            metric=metric,
+            **index_config,
+        )
+        return AsyncCollectionWrapper(col)
+
+    async def get_collection(self, name: str) -> AsyncCollection:
+        """Return an existing collection; raise ``KeyError`` if absent."""
+        col = await asyncio.to_thread(self._sync.get_collection, name)
+        return AsyncCollectionWrapper(col)
+
+    async def get_or_create_collection(
+        self,
+        name: str,
+        *,
+        dimension: Optional[int] = None,
+        metric: str = "cosine",
+        **index_config,
+    ) -> AsyncCollection:
+        """Return collection ``name``, creating it if missing."""
+        col = await asyncio.to_thread(
+            self._sync.get_or_create_collection,
+            name,
+            dimension=dimension,
+            metric=metric,
+            **index_config,
+        )
+        return AsyncCollectionWrapper(col)
+
+    async def delete_collection(self, name: str) -> None:
+        """Drop a collection; raise ``KeyError`` if absent."""
+        await asyncio.to_thread(self._sync.delete_collection, name)
+
+    async def list_collections(self) -> AsyncIterator[str]:
+        """Yield collection names."""
+        names = await asyncio.to_thread(lambda: list(self._sync.list_collections()))
+        for name in names:
+            yield name
+
+    # ----- lifecycle / context manager ------------------------------------ #
+
+    async def close(self) -> None:
+        """Release backend resources. Calls ``close()`` on the sync client if present."""
+        close = getattr(self._sync, "close", None)
+        if close is not None:
+            await asyncio.to_thread(close)
+
+    async def __aenter__(self) -> "AsyncClientWrapper":
+        return self
+
+    async def __aexit__(self, *exc) -> None:
+        await self.close()
+
+
+# --------------------------------------------------------------------------- #
+# Entry point
+# --------------------------------------------------------------------------- #
+
+
+async def connect_async(backend: str, **kwargs) -> AsyncClient:
+    """
+    Async sibling of :func:`vd.connect`.
+
+    Returns an :class:`~vd.AsyncClient`. Today every backend goes through
+    the universal :class:`AsyncClientWrapper` (built on
+    :func:`asyncio.to_thread`); Phase 2 follow-ups will plug in native async
+    clients per backend, which :func:`connect_async` will return instead.
+
+    Parameters
+    ----------
+    backend : str
+        Backend name — same vocabulary as :func:`vd.connect`.
+    **kwargs
+        Forwarded to :func:`vd.connect`.
+
+    Returns
+    -------
+    AsyncClient
+        A live async client. ``await`` once at session start::
+
+            client = await vd.connect_async("memory")
+
+    Examples
+    --------
+    >>> import asyncio, vd
+    >>> async def go():
+    ...     client = await vd.connect_async("memory")
+    ...     col = await client.create_collection("docs", dimension=2)
+    ...     await col.set("a", vd.Document(id="a", text="x", vector=[1.0, 0.0]))
+    ...     return await col.count()
+    >>> asyncio.run(go())
+    1
+    """
+    # Late import to avoid a top-level cycle (vd.util imports nothing in here,
+    # but keep it lazy so this module is safe to import standalone).
+    from vd.util import connect
+
+    # Per-backend native async adapters can be wired here in Phase 2 by
+    # checking a registry for an async constructor before falling back. For
+    # Phase 1 every backend uses the universal wrapper.
+    sync_client = await asyncio.to_thread(connect, backend, **kwargs)
+    return AsyncClientWrapper(sync_client)
+
+
+# --------------------------------------------------------------------------- #
+# hybrid_search_async — async sibling of vd.hybrid_search
+# --------------------------------------------------------------------------- #
+
+
+async def hybrid_search_async(
+    collection: AsyncCollection,
+    query: Union[str, Vector],
+    *,
+    query_text: Optional[str] = None,
+    limit: int = 10,
+    filter: Optional[Filter] = None,
+    k_dense: Optional[int] = None,
+    k_lexical: Optional[int] = None,
+    rrf_k: int = 60,
+    lexical_search: Optional[Callable[..., list[SearchResult]]] = None,
+    egress: Optional[Callable[[SearchResult], Any]] = None,
+    **kwargs,
+) -> AsyncIterator[SearchResult]:
+    """
+    Async sibling of :func:`vd.hybrid_search`.
+
+    If the wrapped sync collection's class supports native hybrid (i.e.
+    satisfies :class:`~vd.SupportsHybrid`), dispatches the whole fused call
+    to a worker thread. Otherwise runs the universal client-side BM25 + RRF
+    fallback in a worker thread too. In both cases the awaitable + async
+    iterator interface stays uniform.
+
+    Parameters mirror :func:`vd.hybrid_search` exactly; see that function for
+    the full docs.
+
+    Yields
+    ------
+    dict
+        Fused result dicts.
+
+    Examples
+    --------
+    >>> import asyncio, vd
+    >>> async def go():
+    ...     client = await vd.connect_async("memory")
+    ...     col = await client.create_collection("docs", dimension=2)
+    ...     await col.set("a", vd.Document(id="a", text="cats",
+    ...                                    vector=[1.0, 0.0]))
+    ...     await col.set("b", vd.Document(id="b", text="dogs",
+    ...                                    vector=[0.0, 1.0]))
+    ...     hits = []
+    ...     async for h in vd.hybrid_search_async(col, [0.9, 0.1],
+    ...                                           query_text="cats", limit=1):
+    ...         hits.append(h["id"])
+    ...     return hits
+    >>> asyncio.run(go())
+    ['a']
+    """
+    from vd.search import hybrid_search as sync_hybrid_search
+
+    sync_collection = getattr(collection, "sync", collection)
+
+    def _run() -> list[SearchResult]:
+        return list(
+            sync_hybrid_search(
+                sync_collection,
+                query,
+                query_text=query_text,
+                limit=limit,
+                filter=filter,
+                k_dense=k_dense,
+                k_lexical=k_lexical,
+                rrf_k=rrf_k,
+                lexical_search=lexical_search,
+                egress=egress,
+                **kwargs,
+            )
+        )
+
+    results = await asyncio.to_thread(_run)
+    for hit in results:
+        yield hit
+
+
+# Re-export SupportsHybrid so users importing from vd.asynchronous have the
+# whole hybrid surface in one place — even if their native-async adapter
+# decides to also satisfy SupportsHybrid directly.
+__all__ = [
+    "AsyncCollectionWrapper",
+    "AsyncClientWrapper",
+    "connect_async",
+    "hybrid_search_async",
+    "SupportsHybrid",
+]
diff --git a/vd/base.py b/vd/base.py
index 522324f..4c84025 100644
--- a/vd/base.py
+++ b/vd/base.py
@@ -34,6 +34,7 @@
 from dataclasses import dataclass, field
 from typing import (
     Any,
+    AsyncIterator,
     Callable,
     Iterable,
     Iterator,
@@ -342,6 +343,95 @@ def hybrid_search(
     ) -> Iterator[SearchResult]: ...
 
 
+# --------------------------------------------------------------------------- #
+# Async protocols — opt-in, every backend covered by the default wrapper
+# --------------------------------------------------------------------------- #
+
+
+@runtime_checkable
+class AsyncCollection(Protocol):
+    """
+    The async sibling of :class:`Collection`.
+
+    Same conceptual surface — storage + ``search`` — but every method is
+    awaitable and iterators are :class:`~typing.AsyncIterator`. The mapping
+    interface is exposed as explicit ``get`` / ``set`` / ``delete`` / ``keys``
+    / ``count`` methods (the stdlib's ``MutableMapping`` ABC has no async
+    counterpart; explicit methods are the Motor / aiopg convention).
+
+    Construct via :func:`vd.connect_async`; the universal
+    :class:`AsyncCollectionWrapper` in :mod:`vd.asynchronous` adapts every
+    backend to this protocol by dispatching to the sync API through
+    :func:`asyncio.to_thread`. Backends with native async SDKs override the
+    wrapper and additionally satisfy :class:`SupportsNativeAsync`.
+    """
+
+    async def get(self, key: str) -> "Document": ...
+
+    async def set(
+        self, key: str, value: Union[str, tuple, "Document"]
+    ) -> None: ...
+
+    async def delete(self, key: str) -> None: ...
+
+    def keys(self) -> AsyncIterator[str]: ...
+
+    async def count(self) -> int: ...
+
+    def search(
+        self,
+        query: Union[str, Vector],
+        *,
+        limit: int = 10,
+        filter: Optional[Filter] = None,
+        egress: Optional[Callable[[SearchResult], Any]] = None,
+        **kwargs,
+    ) -> AsyncIterator[SearchResult]: ...
+
+
+@runtime_checkable
+class AsyncClient(Protocol):
+    """
+    The async sibling of :class:`Client`.
+
+    Same operations — collection create / fetch / drop / list — exposed as
+    awaitables and async iterators. Construct via :func:`vd.connect_async`.
+    """
+
+    async def create_collection(
+        self,
+        name: str,
+        *,
+        dimension: Optional[int] = None,
+        metric: str = "cosine",
+        **index_config,
+    ) -> AsyncCollection: ...
+
+    async def get_collection(self, name: str) -> AsyncCollection: ...
+
+    async def delete_collection(self, name: str) -> None: ...
+
+    def list_collections(self) -> AsyncIterator[str]: ...
+
+
+@runtime_checkable
+class SupportsNativeAsync(Protocol):
+    """
+    Marker protocol set on async clients/collections that use a backend's
+    native async SDK rather than the universal :func:`asyncio.to_thread`
+    wrapper.
+
+    Why care: in high-concurrency event-loop apps (FastAPI, Starlette, etc.),
+    a ``to_thread``-wrapped backend still blocks a worker thread per request.
+    For real non-blocking I/O, prefer collections that satisfy this protocol.
+    The wrapper sets this attribute to ``False``; native adapters set it to
+    ``True``. ``isinstance(c, SupportsNativeAsync)`` matches both — check
+    ``c.native_async`` for the boolean.
+    """
+
+    native_async: bool
+
+
 # --------------------------------------------------------------------------- #
 # Helpers shared by the abstract bases
 # --------------------------------------------------------------------------- #