diff --git a/README.md b/README.md
index 25823f3..15f8063 100644
--- a/README.md
+++ b/README.md
@@ -111,6 +111,59 @@ vc = VeeamClient(
 await vc.connect()
 ```
 
+#### Detect the API version an appliance serves
+
+The REST API has no endpoint that reports its supported versions, and nothing negotiates one
+for you — the version is part of every path, so a client picks one up front. That path is
+also what makes detection possible: `/api/v{version}/system/about` requires a token, so an
+anonymous probe gets 401 where the version is served and 404 where it is not.
+`detect_api_version` returns the newest version that both the appliance serves and this
+library can speak:
+
+```python
+import asyncio
+from veeam_az import VeeamClient
+from veeam_az.discovery import detect_api_version
+
+async def main():
+    base_url = "https://vbaz.example.com"
+
+    api_version = await detect_api_version(base_url, verify_ssl=False)
+    if api_version is None:
+        # The appliance may be unreachable or behind a proxy — choose your own default
+        api_version = "8.1"
+
+    vc = VeeamClient(
+        host=base_url,
+        username="administrator",
+        password="SuperSecretPassword",
+        api_version=api_version,
+        verify_ssl=False,
+    )
+    await vc.connect()
+
+asyncio.run(main())
+```
+
+Detection needs no credentials, so it can run before you have any. Probes are concurrent, so
+it costs roughly one round trip regardless of how many versions this library supports.
+
+If the appliance is not on the usual HTTPS port, `detect_rest_api` finds the port and the
+version at once:
+
+```python
+from veeam_az.discovery import detect_rest_api
+
+endpoint = await detect_rest_api("vbaz.example.com", ports=(443, 8443), verify_ssl=False)
+if endpoint:
+    print(endpoint.port, endpoint.api_version)  # e.g. 443 8.1
+    base_url = f"https://vbaz.example.com{endpoint.base_url_suffix}"
+```
+
+Resolve it once and store the result rather than detecting on every start: an appliance
+upgrade would otherwise silently move you onto a newer version, and versions rename enum
+values and add required fields.
+
 #### Call an API endpoint (async)
 ```python
 policies = await vc.call(
diff --git a/pyproject.toml b/pyproject.toml
index bb90180..6fc73df 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -37,6 +37,7 @@ classifiers = [
 [project.optional-dependencies]
 dev = [
     "pytest>=7.0.0",
+    "pytest-asyncio>=0.23.0",
     "pytest-cov",
     "black",
     "ruff",
@@ -55,4 +56,9 @@ tag_regex = "^(?:v)?(?P<version>.*)$"
 
 [tool.setuptools.packages.find]
 include = ["veeam_az*"]
-exclude = ["media*", "openapi_schemas*"]
+exclude = ["media*", "openapi_schemas*", "tests*"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+asyncio_mode = "strict"
+asyncio_default_fixture_loop_scope = "function"
diff --git a/tests/test_discovery.py b/tests/test_discovery.py
new file mode 100644
index 0000000..a483197
--- /dev/null
+++ b/tests/test_discovery.py
@@ -0,0 +1,280 @@
+"""Tests for API version detection.
+
+Served against httpx.MockTransport, so the probing behaviour is exercised for real rather
+than mocked out.
+"""
+
+import httpx
+import pytest
+
+from veeam_az.discovery import (
+    RestApiEndpoint,
+    detect_api_version,
+    detect_rest_api,
+    newest_first,
+    probe_url,
+)
+from veeam_az.versions import VERSION_TO_PACKAGE
+
+BASE_URL = "https://vbaz.example.com"
+
+
+def make_client(served, status_for_unserved=404, served_status=401, fail_with=None):
+    """An httpx client whose appliance routes only the given versions.
+
+    A served version answers 401 by default, which is what an unauthenticated probe of a
+    real appliance gets: the route exists and the bearer token is missing.
+    """
+    requested = []
+
+    def handler(request):
+        requested.append(str(request.url))
+        if fail_with is not None:
+            raise fail_with
+        for version in served:
+            if request.url.path == f"/api/v{version}/system/about":
+                return httpx.Response(served_status)
+        return httpx.Response(status_for_unserved)
+
+    return httpx.AsyncClient(transport=httpx.MockTransport(handler)), requested
+
+
+def test_probe_url_adds_the_v_prefix():
+    """API paths carry a "v" that the version strings do not."""
+    assert probe_url(BASE_URL, "8.1") == f"{BASE_URL}/api/v8.1/system/about"
+
+
+def test_probe_url_tolerates_a_trailing_slash():
+    assert probe_url(BASE_URL + "/", "8.1").count("//") == 1
+
+
+@pytest.mark.parametrize(
+    "versions,expected",
+    [
+        (["8.1", "9.0"], ["9.0", "8.1"]),
+        # Numeric ordering, which string sorting would get wrong
+        (["8.9", "8.10"], ["8.10", "8.9"]),
+        # Differing depths compare as if padded with zeroes rather than raising
+        (["8", "8.1"], ["8.1", "8"]),
+        # Unrecognizable entries are dropped rather than ordered arbitrarily
+        (["8.1", "nonsense", None, ""], ["8.1"]),
+        ([], []),
+    ],
+)
+def test_newest_first(versions, expected):
+    assert newest_first(versions) == expected
+
+
+@pytest.mark.asyncio
+async def test_detects_the_version_the_appliance_serves():
+    client, _ = make_client(served=list(VERSION_TO_PACKAGE))
+
+    detected = await detect_api_version(BASE_URL, client=client)
+
+    assert detected == newest_first(VERSION_TO_PACKAGE)[0]
+
+
+@pytest.mark.asyncio
+async def test_detects_the_newest_of_several():
+    """An older appliance must not be handed a newer version."""
+    client, _ = make_client(served=["8.1"])
+
+    detected = await detect_api_version(BASE_URL, client=client, versions=["8.1", "9.0"])
+
+    assert detected == "8.1"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("status", [200, 401, 403])
+async def test_any_routed_status_counts_as_served(status):
+    """The question is whether the path routes, not whether the probe was allowed."""
+    client, _ = make_client(served=["8.1"], served_status=status)
+
+    assert await detect_api_version(BASE_URL, client=client) == "8.1"
+
+
+@pytest.mark.asyncio
+async def test_probes_only_versions_this_package_supports():
+    """Reporting a version the SDK cannot speak would be useless to the caller."""
+    client, requested = make_client(served=list(VERSION_TO_PACKAGE))
+
+    await detect_api_version(BASE_URL, client=client)
+
+    probed = {url.split("/api/v")[1].split("/")[0] for url in requested}
+    assert probed == set(VERSION_TO_PACKAGE)
+
+
+@pytest.mark.asyncio
+async def test_returns_none_when_no_version_answers():
+    """The caller then falls back to a version of its own choosing."""
+    client, _ = make_client(served=[])
+
+    assert await detect_api_version(BASE_URL, client=client) is None
+
+
+@pytest.mark.asyncio
+async def test_returns_none_when_the_appliance_is_unreachable():
+    client, _ = make_client(served=[], fail_with=httpx.ConnectError("unreachable"))
+
+    assert await detect_api_version(BASE_URL, client=client) is None
+
+
+@pytest.mark.asyncio
+async def test_returns_none_when_a_probe_times_out():
+    client, _ = make_client(served=[], fail_with=httpx.ReadTimeout("too slow"))
+
+    assert await detect_api_version(BASE_URL, client=client) is None
+
+
+@pytest.mark.asyncio
+async def test_a_single_failing_probe_does_not_hide_the_others():
+    """One version timing out must not lose a version that did answer."""
+
+    def handler(request):
+        if request.url.path == "/api/v9.0/system/about":
+            raise httpx.ReadTimeout("too slow")
+        if request.url.path == "/api/v8.1/system/about":
+            return httpx.Response(401)
+        return httpx.Response(404)
+
+    client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
+
+    detected = await detect_api_version(BASE_URL, client=client, versions=["9.0", "8.1"])
+
+    assert detected == "8.1"
+
+
+@pytest.mark.asyncio
+async def test_empty_candidate_list_makes_no_requests():
+    client, requested = make_client(served=[])
+
+    assert await detect_api_version(BASE_URL, client=client, versions=[]) is None
+    assert requested == []
+
+
+@pytest.mark.asyncio
+async def test_caller_supplied_client_is_left_open():
+    """Reusing a caller's client must not close it out from under them."""
+    client, _ = make_client(served=list(VERSION_TO_PACKAGE))
+
+    await detect_api_version(BASE_URL, client=client)
+
+    assert not client.is_closed
+    await client.aclose()
+
+
+@pytest.mark.asyncio
+async def test_detected_version_is_usable_with_veeam_client():
+    """Detection is only useful if its result routes to a real SDK package."""
+    from veeam_az.client import VeeamClient
+
+    client, _ = make_client(served=list(VERSION_TO_PACKAGE))
+    detected = await detect_api_version(BASE_URL, client=client)
+
+    vc = VeeamClient(host=BASE_URL, api_version=detected, token="token")
+    assert vc.package == VERSION_TO_PACKAGE[detected]
+
+
+# ---------------------------------------------------------------------------
+# Endpoint detection: which port, and which version on it
+# ---------------------------------------------------------------------------
+
+
+def make_endpoint_client(served, fail_with=None):
+    """A client whose appliance routes only the given (port, version) pairs.
+
+    httpx reports url.port as None for a scheme's default port, so an https URL written as
+    ":443" arrives here with no port at all — hence the fallback. A real appliance still
+    sees the TCP port it was reached on.
+    """
+    requested = []
+
+    def port_of(request):
+        return request.url.port or 443
+
+    def handler(request):
+        requested.append((port_of(request), str(request.url.path)))
+        if fail_with is not None:
+            raise fail_with
+        for port, version in served:
+            if port_of(request) == port and request.url.path == f"/api/v{version}/system/about":
+                return httpx.Response(401)
+        return httpx.Response(404)
+
+    return httpx.AsyncClient(transport=httpx.MockTransport(handler)), requested
+
+
+@pytest.mark.asyncio
+async def test_finds_the_default_port():
+    client, _ = make_endpoint_client([(443, "8.1")])
+
+    endpoint = await detect_rest_api("vbaz.example.com", client=client)
+
+    assert endpoint == RestApiEndpoint(port=443, api_version="8.1")
+
+
+@pytest.mark.asyncio
+async def test_port_order_is_the_callers_choice():
+    """A deployment behind a port-forwarding proxy can be anywhere."""
+    client, _ = make_endpoint_client([(443, "8.1"), (8443, "8.1")])
+
+    endpoint = await detect_rest_api("vbaz.example.com", ports=(8443, 443), client=client)
+
+    assert endpoint.port == 8443
+
+
+@pytest.mark.asyncio
+async def test_probes_every_port_and_version_combination():
+    client, requested = make_endpoint_client([])
+
+    await detect_rest_api(
+        "vbaz.example.com", ports=(443, 8443), client=client, versions=["8.1", "9.0"]
+    )
+
+    assert sorted(requested) == sorted(
+        [
+            (port, f"/api/v{version}/system/about")
+            for port in (443, 8443)
+            for version in ("8.1", "9.0")
+        ]
+    )
+
+
+@pytest.mark.asyncio
+async def test_returns_none_when_no_port_answers():
+    """Caller keeps whatever the user configured rather than guessing."""
+    client, _ = make_endpoint_client([])
+
+    assert await detect_rest_api("vbaz.example.com", client=client) is None
+
+
+@pytest.mark.asyncio
+async def test_returns_none_when_the_host_is_unreachable():
+    client, _ = make_endpoint_client([], fail_with=httpx.ConnectError("no route"))
+
+    assert await detect_rest_api("vbaz.example.com", client=client) is None
+
+
+@pytest.mark.asyncio
+async def test_no_ports_means_no_requests():
+    client, requested = make_endpoint_client([])
+
+    assert await detect_rest_api("vbaz.example.com", ports=(), client=client) is None
+    assert requested == []
+
+
+@pytest.mark.asyncio
+async def test_detected_endpoint_builds_a_working_base_url():
+    """The result should drop straight into a VeeamClient host argument."""
+    from veeam_az.client import VeeamClient
+
+    client, _ = make_endpoint_client([(443, "8.1")])
+    endpoint = await detect_rest_api("vbaz.example.com", client=client)
+
+    vc = VeeamClient(
+        host=f"https://vbaz.example.com{endpoint.base_url_suffix}",
+        api_version=endpoint.api_version,
+        token="token",
+    )
+    assert vc.host == "https://vbaz.example.com:443"
+    assert vc.package == VERSION_TO_PACKAGE[endpoint.api_version]
diff --git a/veeam_az/discovery.py b/veeam_az/discovery.py
new file mode 100644
index 0000000..357f18a
--- /dev/null
+++ b/veeam_az/discovery.py
@@ -0,0 +1,240 @@
+"""Detect which REST API version a Veeam Backup for Microsoft Azure appliance offers.
+
+The REST API has no endpoint that reports the set of versions an appliance supports, and
+nothing negotiates it for the caller — the version is part of every path, so a client picks
+one up front and lives with it.
+
+That path is what makes detection possible without credentials. ``/api/v8.1/system/about``
+requires a bearer token, so an anonymous probe distinguishes the two cases cleanly:
+
+* the appliance serves that version — routing matches and authentication rejects it: 401
+  (or 403, or 200 on a deployment that does not gate it)
+* the appliance does not serve that version — routing finds nothing: 404
+
+``detect_api_version`` intersects the versions that answer with the versions this package
+can actually speak (``VERSION_TO_PACKAGE``) and returns the newest, so the answer is always
+something the caller can pass straight to ``VeeamClient``.
+
+Detection is best-effort by contract. An appliance can be unreachable, behind a proxy that
+rewrites statuses, or reject the probe for a reason of its own, and all of those return None
+so a caller falls back to a version of its own choosing rather than failing outright.
+
+Callers should resolve once and store the result. Re-detecting on every start would silently
+move an existing deployment onto a newer version, and versions rename enum values and add
+required fields.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from collections.abc import Iterable, Sequence
+from typing import NamedTuple
+
+import httpx
+
+from .versions import VERSION_TO_PACKAGE
+
+_LOGGER = logging.getLogger(__name__)
+
+# Probes run concurrently, so this bounds the whole detection rather than each version
+DEFAULT_TIMEOUT = 8.0
+
+# Requires a bearer token and exists in every version this package ships, so an anonymous
+# request separates "this version is served" from "this version is not" by status alone.
+# The path carries a "v" prefix that the version strings themselves do not.
+PROBE_PATH = "/api/v{version}/system/about"
+
+# Statuses that mean the route exists. 401 is the expected answer to an unauthenticated
+# probe; 200 and 403 also prove the path is routed, which is the only question being asked.
+# Anything else — 404 above all — means this version is not served here.
+SERVED_STATUSES = frozenset({200, 401, 403})
+
+# The appliance publishes its web UI and REST API on HTTPS
+REST_PORT = 443
+DEFAULT_PORTS = (REST_PORT,)
+
+
+class RestApiEndpoint(NamedTuple):
+    """Where an appliance's REST API answers."""
+
+    port: int
+    api_version: str
+
+    @property
+    def base_url_suffix(self) -> str:
+        """Port suffix for building a base URL, e.g. ":443"."""
+        return f":{self.port}"
+
+
+def probe_url(base_url: str, version: str) -> str:
+    """Build the URL this module probes for one API version."""
+    return f"{base_url.rstrip('/')}{PROBE_PATH.format(version=version)}"
+
+
+def newest_first(versions: Iterable[str]) -> list[str]:
+    """Order API versions newest first, dropping any that are not recognizable.
+
+    Versions look like "8.1". Comparing the numbers rather than the strings keeps a
+    hypothetical "8.10" above "8.9" instead of below it.
+    """
+
+    def key(version):
+        try:
+            parts = version.lstrip("vV").split(".")
+            return tuple(int(part) for part in parts)
+        except (AttributeError, TypeError, ValueError):
+            return None
+
+    # Filter before sorting: an unrecognizable entry has no key to compare, and mixing
+    # those into the sort raises rather than just ordering them last. Pad to a common
+    # length so "8" and "8.1" compare as (8, 0) and (8, 1) rather than raising.
+    ranked = [(key(version), version) for version in versions]
+    ranked = [entry for entry in ranked if entry[0]]
+    width = max((len(key) for key, _ in ranked), default=0)
+    return [
+        version
+        for _, version in sorted(
+            ((key + (0,) * (width - len(key)), version) for key, version in ranked),
+            reverse=True,
+        )
+    ]
+
+
+async def _serves(client, base_url: str, version: str, timeout: float):
+    """Return the version if the appliance routes it, else None."""
+    url = probe_url(base_url, version)
+    try:
+        response = await client.get(url, timeout=timeout)
+    except Exception as err:
+        _LOGGER.debug("%s did not answer: %s", url, err)
+        return None
+
+    if response.status_code in SERVED_STATUSES:
+        return version
+
+    _LOGGER.debug("%s returned HTTP %s", url, response.status_code)
+    return None
+
+
+async def detect_api_version(
+    base_url: str,
+    *,
+    verify_ssl: bool = True,
+    timeout: float = DEFAULT_TIMEOUT,
+    versions: Sequence[str] | None = None,
+    client: httpx.AsyncClient | None = None,
+) -> str | None:
+    """Return the newest API version this appliance serves and this package supports.
+
+    Args:
+        base_url: Appliance base URL, e.g. "https://vbaz.example.com".
+        verify_ssl: Whether to verify the server certificate. Ignored when ``client`` is
+            given, since the caller's client carries its own settings.
+        timeout: Per-request timeout in seconds. Probes are concurrent.
+        versions: Candidate versions. Defaults to everything this package can speak.
+        client: An existing httpx.AsyncClient to reuse instead of opening one.
+
+    Returns:
+        A version string such as "8.1", or None if nothing answered — the caller should then
+        fall back to a version it chooses.
+    """
+    candidates = newest_first(VERSION_TO_PACKAGE if versions is None else versions)
+    if not candidates:
+        return None
+
+    owns_client = client is None
+    if owns_client:
+        client = httpx.AsyncClient(verify=verify_ssl)
+
+    try:
+        results = await asyncio.gather(
+            *(_serves(client, base_url, version, timeout) for version in candidates),
+            return_exceptions=True,
+        )
+    finally:
+        if owns_client:
+            await client.aclose()
+
+    served = {result for result in results if isinstance(result, str)}
+    if not served:
+        _LOGGER.debug("No API version answered on %s", base_url)
+        return None
+
+    # candidates is newest-first, so the first match is the newest version served
+    detected = next(version for version in candidates if version in served)
+    _LOGGER.debug("%s serves %s; selected %s", base_url, sorted(served), detected)
+    return detected
+
+
+async def detect_rest_api(
+    host: str,
+    *,
+    ports: Sequence[int] = DEFAULT_PORTS,
+    versions: Sequence[str] | None = None,
+    verify_ssl: bool = True,
+    timeout: float = DEFAULT_TIMEOUT,
+    client: httpx.AsyncClient | None = None,
+) -> RestApiEndpoint | None:
+    """Find where an appliance answers: which port, and which API version.
+
+    The appliance serves the REST API on HTTPS alongside its web UI, so 443 is the only
+    candidate worth trying by default — but a deployment behind a port-forwarding proxy can
+    be anywhere, and the probe URL contains the port, so one sweep answers both questions at
+    once rather than detecting a port and then re-probing for a version.
+
+    Args:
+        host: Hostname or address, without scheme or port.
+        ports: Candidate ports, in order of preference. The first that answers wins, so a
+            caller with a non-default deployment should list its own port first.
+        versions: Candidate versions. Defaults to everything this package can speak.
+        verify_ssl: Whether to verify the server certificate. Ignored when ``client`` is
+            given, since the caller's client carries its own settings.
+        timeout: Per-request timeout in seconds. Probes are concurrent.
+        client: An existing httpx.AsyncClient to reuse instead of opening one.
+
+    Returns:
+        A RestApiEndpoint(port, api_version), or None if nothing answered.
+    """
+    candidates = newest_first(VERSION_TO_PACKAGE if versions is None else versions)
+    if not candidates or not ports:
+        return None
+
+    owns_client = client is None
+    if owns_client:
+        client = httpx.AsyncClient(verify=verify_ssl)
+
+    attempts = [(port, version) for port in ports for version in candidates]
+
+    try:
+        results = await asyncio.gather(
+            *(
+                _serves(client, f"https://{host}:{port}", version, timeout)
+                for port, version in attempts
+            ),
+            return_exceptions=True,
+        )
+    finally:
+        if owns_client:
+            await client.aclose()
+
+    answered = {
+        (port, version)
+        for (port, version), result in zip(attempts, results)
+        if isinstance(result, str)
+    }
+    if not answered:
+        _LOGGER.debug("Nothing answered on %s across ports %s", host, list(ports))
+        return None
+
+    # ports is preference-ordered and candidates is newest-first, so the first hit in that
+    # nesting is the preferred port running its newest served version
+    port, version = next(attempt for attempt in attempts if attempt in answered)
+    _LOGGER.debug(
+        "%s answered on %s; selected port %s with %s",
+        host,
+        sorted(answered),
+        port,
+        version,
+    )
+    return RestApiEndpoint(port=port, api_version=version)