python-fleet-telemetry

An async Python library that receives Tesla Fleet Telemetry directly from vehicles. It terminates the mutual-TLS WebSocket connection in-process, decodes the FlatBuffers transport envelope and the protobuf payloads, and exposes each frame as a typed Record through event listeners or an async record stream.

This is the receive side — the mirror image of the Teslemetry relay consumed by python-teslemetry-stream. Where that library is a client of a relay, this library is the server that vehicles connect to. Python 3.13+.

How it fits

Tesla vehicles can be configured to stream telemetry over an mTLS WebSocket to a server you run. The reference implementation is the Go tesla-fleet-telemetry server. python-fleet-telemetry plays the same role: it is the endpoint vehicles connect to, not a consumer of a downstream message bus.

In scope: accepting vehicle connections, terminating mTLS, deriving the VIN from the verified client certificate, decoding envelopes and payloads, tracking live connections, and delivering records to listeners and streams.

Out of scope (use python-tesla-fleet-api): application registration, partner tokens, virtual-key pairing, pushing the fleet_telemetry_config that tells vehicles where to stream, and issuing the TLS/CA material. This library assumes those are already in place.

Install

pip install python-fleet-telemetry

Not yet published to PyPI — this is the intended install command.

Runtime dependencies: aiohttp, protobuf>=6.32, flatbuffers, cryptography. Requires Python 3.13+.

Prerequisites

Before this server can receive anything:

• Your vehicles must be configured (via the Fleet API — out of scope here, see python-tesla-fleet-api) to stream to your server's fully-qualified domain name.
• You need a TLS server certificate for that FQDN and its private key.
• You need the Tesla CA bundle used to verify the client certificates the vehicles present.

mTLS is mandatory. The VIN of each connection is derived from the verified client certificate's subject Common Name — there is no other authentication path. Client certificates must be issued by a recognized Tesla CA; a certificate from an unrecognized issuer is rejected with HTTP 496 before the WebSocket upgrade. The TelemetryServer constructor enforces this by requiring ssl_context.verify_mode == ssl.CERT_REQUIRED, raising ValueError otherwise.

Certificates

The mTLS here has two independent trust directions, and fleet_telemetry.certs handles both so you don't have to touch openssl:

• The vehicle verifies your server. Your server presents a TLS leaf certificate for its FQDN, chained to a CA the vehicle trusts. You register that CA's certificate as the ca field of the vehicle's fleet_telemetry_config. A vehicle rejects a bare self-signed cert used directly, so it must be a CA → leaf chain.
• Your server verifies the vehicle. The vehicle presents a client certificate issued by Tesla's CA (fixed, not generated by you). Your server must trust Tesla's CA bundle to complete the handshake — this package vendors it.

ServerCredentials generates and persists a root CA plus a server leaf under a directory you choose (get-or-create, like python-tesla-fleet-api's key handling), and builds a ready-to-use SSLContext:

from fleet_telemetry import ServerCredentials, TelemetryServer

creds = ServerCredentials("/config/fleet_telemetry")   # persistent directory
creds.ensure("telemetry.example.com")                  # create/load CA + server leaf

# Register creds.ca_certificate_pem as fleet_telemetry_config.ca (via
# python-tesla-fleet-api) so the vehicle trusts your server. The CA is long-lived;
# rotate the leaf with creds.rotate_server_cert(fqdn) WITHOUT re-pushing config.

server = TelemetryServer(ssl_context=creds.build_ssl_context(), port=443)

Private keys are written 0600 in a 0700 directory. build_ssl_context() trusts Tesla's production CA bundle by default, or the engineering/staging bundle with staging=True. Cert generation does blocking key generation and disk I/O — inside an event loop (e.g. Home Assistant), run ensure() / rotate_server_cert() in an executor. Deleting the directory is the deliberate way to regenerate the CA (a partial/half-deleted state raises rather than silently minting a new trust anchor, which would invalidate every vehicle's registered ca).

You may still supply your own SSLContext (e.g. with a publicly-trusted server cert) instead of ServerCredentials — the quickstart below shows the manual form.

Quickstart

import asyncio
import ssl

from fleet_telemetry import TelemetryServer, Topic


def build_ssl_context() -> ssl.SSLContext:
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
    ctx.load_cert_chain("server-cert.pem", "server-key.pem")
    ctx.load_verify_locations("tesla-ca.pem")
    ctx.verify_mode = ssl.CERT_REQUIRED
    return ctx


async def main() -> None:
    server = TelemetryServer(ssl_context=build_ssl_context(), port=443)

    # Callback style
    server.on_data(lambda r: print(r.vin, r.fields()))
    server.on_connectivity(lambda r: print("connectivity:", r.vin, r.topic))

    async with server:
        print("listening for vehicles on :443")

        # Stream style: iterate records as they arrive
        async with server.records() as stream:
            async for record in stream:
                print(record.vin, record.topic, record.created_at)


if __name__ == "__main__":
    asyncio.run(main())

A complete version is in examples/basic.py.

End-to-end example

examples/end_to_end.py wires up the whole flow — the half basic.py leaves out. It reads its settings from environment variables, gets-or-creates the mTLS certificates with ServerCredentials, pushes a fleet_telemetry_config to the vehicle over the Fleet API (via tesla-fleet-api, registering the generated root CA as ca so the vehicle trusts your server), polls until the vehicle reports synced == true, then starts the server and prints records as they arrive.

Install the extra dependency it needs with the example dependency group:

uv sync --group example

Run it against a fully-onboarded app + paired vehicle (see the module docstring for the complete prerequisite list — developer app, partner token, virtual-key pairing, and a publicly reachable FQDN):

export TESLA_ACCESS_TOKEN=...          # user OAuth token (device-data + command)
export TESLA_VIN=5YJ...                # the vehicle to configure
export TELEMETRY_FQDN=telemetry.example.com   # public hostname the vehicle dials
# optional: TESLA_REGION=na  TELEMETRY_PORT=443  TELEMETRY_CERT_DIR=./fleet_telemetry_certs
uv run python examples/end_to_end.py

Pass --dry-run (or set DRY_RUN=1) to generate the certificates and print the exact config body it would push, then exit without any Fleet API call or listening socket — handy for inspecting the payload without live credentials:

TESLA_ACCESS_TOKEN=x TESLA_VIN=5YJ... TELEMETRY_FQDN=telemetry.example.com \
  uv run python examples/end_to_end.py --dry-run

Edit the DEFAULT_FIELDS mapping at the top of the file to change which signals stream and how often. If you would rather not host the receive server and Fleet API plumbing yourself, the fully-managed alternative is Teslemetry.

The server is an async context manager (async with server: starts on enter and stops gracefully on exit); you can also call await server.start() / await server.stop() directly.

Callbacks may be sync or async and each receives a single Record. Every registration method returns an unsubscribe callable.

By design (reliable-ack=false), the current frame is always acked to the vehicle before any listener runs, so a failing or slow listener never delays or denies the ack for that frame. Listener exceptions are isolated: one raising listener never prevents the other listeners or the records() fan-out from receiving the record.

Backpressure, however, is real and intentional: dispatch awaits async listeners inline within the per-connection read loop, so a slow or hung async listener applies backpressure to processing of subsequent frames on every connection that dispatches to it (listeners are shared across connections). A hung sync listener blocks the whole event loop (inherent to asyncio). Keep listeners fast, or offload heavy work to a task or queue. The records() async-iterator consumers are the exception: they are bounded (drop-oldest) and are not stalled by slow listeners.

server.records() returns an independent async iterator; multiple concurrent iterators each receive every record. Its queue is bounded (queue_maxsize, default 1000) and drops the oldest record under backpressure, so a slow consumer can never stall ingestion.

Filtering

add_listener filters across three independent dimensions — VIN, topic, and signal name (field) — combined with AND semantics. Each accepts a single value, an iterable of values, or None (no constraint on that dimension):

from fleet_telemetry import Topic

server.add_listener(
    lambda r: print(r.vin, r.fields()["VehicleSpeed"]),
    vin="LRW...123",
    topic=Topic.DATA,
    field="VehicleSpeed",
)

A field= constraint fires when the record's fields() contains that signal name. An empty iterable (e.g. vin=[]) is a constraint nothing can satisfy, so the listener never fires.

The topic-preset helpers on_data, on_alert, on_error, and on_connectivity are thin wrappers over add_listener with topic fixed; they accept vin and field.

Live connection state is also available: server.connections returns a dict[str, Connection] snapshot keyed by VIN, and server.is_connected(vin) returns a bool.

Record shape

Every WebSocket frame becomes exactly one frozen Record (never one per signal):

Attribute	Type	Notes
vin	str	Derived from the verified client certificate
topic	Topic	DATA / ALERTS / ERRORS / CONNECTIVITY
created_at	datetime	Timezone-aware UTC
txid	str	Transport transaction id (empty for synthetic records)
message	protobuf message	Payload / VehicleAlerts / VehicleErrors / VehicleConnectivity
raw	bytes	Original payload bytes, for re-decode or verbatim forwarding

Record.fields() returns a dict[str, object] mapping signal name to value — for DATA records only ({} for every other topic). Values come from the Value oneof set on each datum (None if unset).

Forward compatibility: the bundled protobuf schema is a point-in-time snapshot. A vehicle on newer firmware can emit a signal this library doesn't yet know by name; rather than raising, such signals appear in fields() under the synthetic key Field_<int> (the raw enum number). If a payload repeats a field key, the last occurrence wins.

Topic is a StrEnum: DATA="V", ALERTS="alerts", ERRORS="errors", CONNECTIVITY="connectivity". Connectivity records are synthesized by the server as vehicles connect and disconnect (with a CONNECTED / DISCONNECTED ConnectivityEvent status), in addition to any connectivity frames a vehicle sends itself.

Connection is frozen with .vin, .connected_at (tz-aware UTC), .peer (remote address), and .client_version (from the connection's Version header, may be None).

Known limitations and hardening

This library handles connection termination and parsing; it does not yet include production abuse protections. Be honest with yourself about your threat model before exposing it to the open internet:

• No built-in rate limiting. Frames are acked and dispatched with no throttle.
• No per-VIN connection cap.
• No idle / half-open connection timeout by default. A silent or hostile peer holds a connection slot indefinitely.

The available knob is aiohttp's WebSocket receive_timeout / heartbeat, which would drop silent peers — but enabling it needs a chosen policy, so it is not turned on by default. For production fleets, front this server with infrastructure-level protections (load balancer connection limits, WAF, network ACLs, per-source rate limiting).

The bundled protobuf and FlatBuffers schemas are vendored from a point-in-time Tesla reference and regenerated via proto/generate.sh and fbs/generate.sh.

Development

uv sync
uv run pytest          # test suite
uv run ruff check .    # lint
uv run pyright         # type-check (strict)

Regenerate the vendored bindings after updating the schemas in proto/ or fbs/:

./proto/generate.sh    # protobuf bindings -> fleet_telemetry/proto/
./fbs/generate.sh      # FlatBuffers bindings -> fleet_telemetry/_flatbuffers/  (needs flatc)

Roadmap

Possible future directions (not implemented today): typed per-signal listener helpers (e.g. listen_VehicleSpeed(...)), and an opt-in idle-timeout policy.

License

Apache-2.0.