feat(teslemetry): add OAuth dynamic client registration helper

AGENTS.md

@@ -150,6 +150,7 @@ Keep the `tesla-protocol` floor at `>=0.5.0`; earlier releases have generated `.
 - **`_stream_sinks` peels subscription pushes off the command-reply queue before routing**: a `vehicleDataSubscription`'s pushes arrive addressed to us on the same domain queue (`_queues`) an ordinary command's reply uses, correlated by the subscribe request's own `request_uuid`. `_on_message` (`bluetooth.py`) checks `self._stream_sinks.get(msg.request_uuid)` before touching `_queues` — a match routes into that subscription's own bounded, drop-oldest `_StreamSink` instead, so `_send`'s pre-send drain can never discard a push and `_await_response` can never return one as an unrelated command's reply. `_register_stream_sink`/`_unregister_stream_sink` are the only entry points into the registry; there is no public subscription API yet. Tests: `tests/test_ble_stream_sink.py`.
 - **`VehicleAction`/`GetVehicleData` proto coverage is locked by test, not just by convention**: `tests/test_proto_coverage_lock.py` walks both descriptors and fails if any field has no wrapper (`commands.py`) or reader (`bluetooth.py`) and isn't on one of its two small, reasoned allowlists — keep that test in sync with any future `tesla-protocol` bump rather than special-casing new fields elsewhere. The only fields deliberately left unwrapped today are the 7-field push-style subscription/streaming family (`createStreamSession`/`streamMessage`/`vehicleDataSubscription`/`vehicleDataAck`/`vitalsSubscription`/`vitalsAck`/`cancelVehicleDataSubscription`, which need a public lifecycle/iterator API atop the private `_stream_sinks` routing above) and `getVehicleImageState` (needs chunked binary-transfer paging). CarServer's `GetVehicleState` sub-state is exposed as `legacy_vehicle_state()` (`bluetooth.py`), matching the `VehicleData.legacy_vehicle_state` reply field name, to avoid confusion with `vehicle_state()` (VCSEC `VehicleStatus`, a different message/domain). `set_rate_tariff`/`add_managed_charging_site` (`commands.py`) take `tesla_protocol` message types directly for their deeply-nested arguments rather than a parallel flattened dataclass API.
 - **Energy-gateway authorized-client pairing has security- and protocol-specific constraints**: use RSA for LAN TEDapi v1r, treat `PENDING_VERIFICATION_TIMEOUT` as terminal, and account for presence-free key removal. The authoritative pairing, retry, encoding, and removal guidance is in `docs/energy_local_control.md`; enum values and API contracts live in `const.py` and the relevant method docstrings.
+- **`register_client()` (`teslemetry/teslemetry.py`) is Teslemetry-only OAuth Dynamic Client Registration (RFC 7591)**: a module-level function, not a `Teslemetry` instance method, since registration precedes having a `client_id` or access token — callers pass a bare `aiohttp.ClientSession`. It always registers a new client (no dedup/caching) and raises `TeslemetryRegistrationError` (`exceptions.py`) on transport failure, a non-2xx response, a non-JSON body, or a response missing a usable `client_id`; a non-dict-but-valid-JSON body (list/scalar) is treated as the same malformed-response error rather than raising an uncaught `AttributeError`. Fleet API and Tessie have no equivalent — don't add one speculatively. See `docs/teslemetry.md`'s "OAuth Dynamic Client Registration" section and `tests/test_teslemetry_register_client.py`.
 
 ## Maintaining this file
 

docs/teslemetry.md

@@ -645,3 +645,41 @@ async def main():
 
 asyncio.run(main())
 ```
+
+## OAuth Dynamic Client Registration
+
+Teslemetry supports OAuth 2.0 Dynamic Client Registration
+([RFC 7591](https://www.rfc-editor.org/rfc/rfc7591)) so each installation can
+register its own `client_id` instead of every client sharing one hardcoded
+value. `register_client` is a standalone helper - it needs only an
+`aiohttp.ClientSession`, not a `Teslemetry` instance, since no `client_id` or
+access token exists yet at registration time. It always registers a new
+client; callers are responsible for persisting the returned `client_id` and
+skipping registration on subsequent runs.
+
+It raises `tesla_fleet_api.exceptions.TeslemetryRegistrationError` if the
+registration endpoint can't be reached, the response isn't valid JSON, or the
+response doesn't contain a usable `client_id`.
+
+```python
+import aiohttp
+from tesla_fleet_api import register_client
+from tesla_fleet_api.exceptions import TeslemetryRegistrationError
+
+async def main():
+    async with aiohttp.ClientSession() as session:
+        try:
+            registration = await register_client(
+                session,
+                client_name="Home Assistant",
+                software_id="home-assistant",
+                software_version="2026.1.0",
+            )
+        except TeslemetryRegistrationError as e:
+            print(e)
+            return
+
+        print(registration.client_id)
+
+asyncio.run(main())
+```

tesla_fleet_api/__init__.py

@@ -14,7 +14,11 @@
 from tesla_fleet_api.tesla.bluetooth import TeslaBluetooth
 from tesla_fleet_api.tesla.fleet import TeslaFleetApi
 from tesla_fleet_api.tesla.oauth import TeslaFleetOAuth
-from tesla_fleet_api.teslemetry.teslemetry import Teslemetry
+from tesla_fleet_api.teslemetry.teslemetry import (
+    Teslemetry,
+    TeslemetryClientRegistration,
+    register_client,
+)
 from tesla_fleet_api.tessie.tessie import Tessie
 from tesla_fleet_api.util import firmware_at_least, firmware_compare
 
@@ -27,10 +31,12 @@
     "TeslaBluetooth",
     "TeslaFleetOAuth",
     "Teslemetry",
+    "TeslemetryClientRegistration",
     "Tessie",
     "firmware_at_least",
     "firmware_compare",
     "get_tariff_periods",
     "is_valid_region",
+    "register_client",
     "unwrap_tariff_v2",
 ]

tesla_fleet_api/exceptions.py

@@ -205,6 +205,18 @@ class OAuthExpired(TeslaFleetError):
     key = "token expired (401)"
 
 
+class TeslemetryRegistrationError(TeslaFleetError):  # Teslemetry specific
+    """Teslemetry OAuth dynamic client registration (RFC 7591) failed.
+
+    Covers a transport/timeout failure reaching the registration endpoint, a
+    non-2xx response, a response body that isn't valid JSON, and a
+    well-formed response missing a usable ``client_id``. The specific reason
+    is carried in ``data``.
+    """
+
+    message = "Teslemetry dynamic client registration failed."
+
+
 class LoginRequired(TeslaFleetError):  # Native and Teslemetry
     """The user has reset their password and a new auth code is required, or the refresh_token has already been used."""
 

tesla_fleet_api/teslemetry/__init__.py

@@ -1,12 +1,18 @@
 from tesla_fleet_api.tesla.charging import Charging
 from tesla_fleet_api.tesla.energysite import EnergySite, EnergySites
 from tesla_fleet_api.tesla.user import User
-from tesla_fleet_api.teslemetry.teslemetry import Teslemetry
+from tesla_fleet_api.teslemetry.teslemetry import (
+    Teslemetry,
+    TeslemetryClientRegistration,
+    register_client,
+)
 from tesla_fleet_api.teslemetry.vehicle import TeslemetryVehicle as Vehicle
 from tesla_fleet_api.teslemetry.vehicle import TeslemetryVehicles as Vehicles
 
 __all__ = [
     "Teslemetry",
+    "TeslemetryClientRegistration",
+    "register_client",
     "Charging",
     "EnergySites",
     "EnergySite",

tesla_fleet_api/teslemetry/teslemetry.py

@@ -1,14 +1,95 @@
 from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
 from time import time
-from typing import Any
+from typing import Any, Final, cast
 
 import aiohttp
 
 from tesla_fleet_api.const import LOGGER, Method, is_valid_region
+from tesla_fleet_api.exceptions import TeslemetryRegistrationError
 from tesla_fleet_api.tesla import TeslaFleetApi
 from tesla_fleet_api.teslemetry.energysite import TeslemetryEnergySites
 from tesla_fleet_api.teslemetry.vehicle import TeslemetryVehicles
 
+REGISTER_URL: Final = "https://api.teslemetry.com/oauth/register"
+
+
+@dataclass(frozen=True, slots=True)
+class TeslemetryClientRegistration:
+    """Parsed result of :func:`register_client`.
+
+    ``client_id`` is the OAuth client identifier this installation should
+    present for every future authorization, including reauthentication.
+    ``raw`` is the full decoded registration response for anything this
+    wrapper doesn't model.
+    """
+
+    client_id: str
+    raw: dict[str, Any]
+
+
+async def register_client(
+    session: aiohttp.ClientSession,
+    client_name: str,
+    software_id: str,
+    software_version: str,
+) -> TeslemetryClientRegistration:
+    """Dynamically register an OAuth client with Teslemetry (RFC 7591).
+
+    Posts to Teslemetry's client-registration endpoint and parses the
+    result. This is a bare transport call - it always registers a new
+    client and has no knowledge of whether the caller already has one;
+    callers are responsible for persisting the returned ``client_id`` and
+    skipping registration on subsequent runs.
+
+    Args:
+        session: An aiohttp session to issue the request on.
+        client_name: Human-readable client name shown during consent.
+        software_id: Identifier for the calling application.
+        software_version: Version string for the calling application.
+
+    Returns:
+        The parsed registration result.
+
+    Raises:
+        TeslemetryRegistrationError: The endpoint could not be reached, the
+            response was not valid JSON, or the response didn't contain a
+            usable ``client_id``.
+    """
+    try:
+        async with session.post(
+            REGISTER_URL,
+            json={
+                "client_name": client_name,
+                "software_id": software_id,
+                "software_version": software_version,
+            },
+        ) as response:
+            response.raise_for_status()
+            registration = await response.json()
+    except (aiohttp.ClientError, TimeoutError) as err:
+        raise TeslemetryRegistrationError(
+            "Could not reach Teslemetry to register a client",
+            status=getattr(err, "status", None),
+        ) from err
+    except ValueError as err:
+        raise TeslemetryRegistrationError(
+            "Teslemetry returned a malformed registration response"
+        ) from err
+
+    registration_dict = (
+        cast("dict[str, Any]", registration) if isinstance(registration, dict) else None
+    )
+    client_id = registration_dict.get("client_id") if registration_dict else None
+    if not isinstance(client_id, str) or not client_id:
+        raise TeslemetryRegistrationError(
+            "Teslemetry registration response did not contain a client_id"
+        )
+
+    return TeslemetryClientRegistration(
+        client_id=client_id, raw=registration_dict or {}
+    )
+
 
 class Teslemetry(TeslaFleetApi):
     vehicles: TeslemetryVehicles

tests/test_teslemetry_register_client.py

@@ -0,0 +1,151 @@
+"""Tests for ``register_client()``, the Teslemetry OAuth dynamic client
+registration (RFC 7591) transport helper.
+
+Mirrors the registration semantics of the Home Assistant Teslemetry
+integration's ``oauth.py`` (endpoint, request payload, response parsing,
+and failure modes): a transport/timeout failure, a server-rejected
+registration, and a malformed/missing-``client_id`` response all raise
+:class:`~tesla_fleet_api.exceptions.TeslemetryRegistrationError`.
+"""
+
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+from typing import Any
+from unittest import IsolatedAsyncioTestCase
+from unittest.mock import AsyncMock, MagicMock
+
+import aiohttp
+
+from tesla_fleet_api.exceptions import TeslemetryRegistrationError
+from tesla_fleet_api.teslemetry.teslemetry import (
+    REGISTER_URL,
+    TeslemetryClientRegistration,
+    register_client,
+)
+
+
+def _fake_response(
+    *, json_body: object = None, raise_for_status_error: Exception | None = None
+) -> MagicMock:
+    resp = MagicMock()
+    resp.raise_for_status = MagicMock(side_effect=raise_for_status_error)
+    if isinstance(json_body, Exception):
+        resp.json = AsyncMock(side_effect=json_body)
+    else:
+        resp.json = AsyncMock(return_value=json_body)
+    return resp
+
+
+def _make_session(
+    *,
+    response: object = None,
+    post_error: Exception | None = None,
+) -> MagicMock:
+    session = MagicMock()
+
+    if post_error is not None:
+        session.post = MagicMock(side_effect=post_error)
+        return session
+
+    @asynccontextmanager
+    async def _ctx(*args: Any, **kwargs: Any):
+        yield response
+
+    session.post = MagicMock(side_effect=lambda *a, **k: _ctx(*a, **k))
+    return session
+
+
+class RegisterClientSuccessTests(IsolatedAsyncioTestCase):
+    async def test_registers_new_client(self) -> None:
+        session = _make_session(
+            response=_fake_response(json_body={"client_id": "new-client-id"})
+        )
+
+        result = await register_client(
+            session, "Home Assistant", "home-assistant", "2026.1.0"
+        )
+
+        self.assertEqual(
+            result,
+            TeslemetryClientRegistration(
+                client_id="new-client-id", raw={"client_id": "new-client-id"}
+            ),
+        )
+        args, kwargs = session.post.call_args
+        self.assertEqual(args[0], REGISTER_URL)
+        self.assertEqual(
+            kwargs["json"],
+            {
+                "client_name": "Home Assistant",
+                "software_id": "home-assistant",
+                "software_version": "2026.1.0",
+            },
+        )
+
+
+class RegisterClientRejectionTests(IsolatedAsyncioTestCase):
+    async def test_server_rejected_registration_raises_typed_error(self) -> None:
+        error = aiohttp.ClientResponseError(
+            MagicMock(), (), status=400, message="Bad Request"
+        )
+        session = _make_session(
+            response=_fake_response(json_body={}, raise_for_status_error=error)
+        )
+
+        with self.assertRaises(TeslemetryRegistrationError):
+            await register_client(session, "Home Assistant", "home-assistant", "1")
+
+    async def test_connection_error_raises_typed_error(self) -> None:
+        session = _make_session(post_error=aiohttp.ClientConnectionError("boom"))
+
+        with self.assertRaises(TeslemetryRegistrationError):
+            await register_client(session, "Home Assistant", "home-assistant", "1")
+
+    async def test_timeout_raises_typed_error(self) -> None:
+        session = _make_session(post_error=TimeoutError())
+
+        with self.assertRaises(TeslemetryRegistrationError):
+            await register_client(session, "Home Assistant", "home-assistant", "1")
+
+
+class RegisterClientMalformedResponseTests(IsolatedAsyncioTestCase):
+    async def test_non_json_body_raises_typed_error(self) -> None:
+        session = _make_session(
+            response=_fake_response(json_body=ValueError("not json"))
+        )
+
+        with self.assertRaises(TeslemetryRegistrationError):
+            await register_client(session, "Home Assistant", "home-assistant", "1")
+
+    async def test_missing_client_id_raises_typed_error(self) -> None:
+        session = _make_session(response=_fake_response(json_body={}))
+
+        with self.assertRaises(TeslemetryRegistrationError):
+            await register_client(session, "Home Assistant", "home-assistant", "1")
+
+    async def test_non_string_client_id_raises_typed_error(self) -> None:
+        session = _make_session(response=_fake_response(json_body={"client_id": 12345}))
+
+        with self.assertRaises(TeslemetryRegistrationError):
+            await register_client(session, "Home Assistant", "home-assistant", "1")
+
+    async def test_empty_string_client_id_raises_typed_error(self) -> None:
+        session = _make_session(response=_fake_response(json_body={"client_id": ""}))
+
+        with self.assertRaises(TeslemetryRegistrationError):
+            await register_client(session, "Home Assistant", "home-assistant", "1")
+
+    async def test_non_dict_body_raises_typed_error(self) -> None:
+        session = _make_session(
+            response=_fake_response(json_body=["unexpected", "list", "body"])
+        )
+
+        with self.assertRaises(TeslemetryRegistrationError):
+            await register_client(session, "Home Assistant", "home-assistant", "1")
+
+    async def test_null_body_raises_typed_error(self) -> None:
+        session = _make_session(response=_fake_response(json_body=None))
+
+        with self.assertRaises(TeslemetryRegistrationError):
+            await register_client(session, "Home Assistant", "home-assistant", "1")