Tesla Fleet API

Tesla Fleet API is a Python library that provides an interface to interact with Tesla's Fleet API, including signed commands and encrypted local Bluetooth (BLE) communication. It also supports interactions with Teslemetry and Tessie services.

Features

• Fleet API for vehicles
• Fleet API for energy sites
• Fleet API with signed vehicle commands
• Bluetooth for vehicles
• Routing and failover across backends for vehicles and energy sites (e.g. Bluetooth/local primary, cloud fallback)
• Teslemetry integration
• Tessie integration

Installation

You can install the library using pip:

pip install tesla-fleet-api

Usage

Authentication

The TeslaFleetOAuth class provides methods that help with authenticating to the Tesla Fleet API. Here's a basic example:

import asyncio
import aiohttp
from tesla_fleet_api import TeslaFleetOAuth

async def main():
    async with aiohttp.ClientSession() as session:
        oauth = TeslaFleetOAuth(
            session=session,
            client_id="<client_id>",
            client_secret="<client_secret>",
            redirect_uri="<redirect_uri>",
        )

        # Get the login URL and navigate the user to it
        login_url = oauth.get_login_url(scopes=["openid", "email", "offline_access"])
        print(f"Please go to {login_url} and authorize access.")

        # After the user authorizes access, they will be redirected to the redirect_uri with a code
        code = input("Enter the code you received: ")

        # Exchange the code for a refresh token
        await oauth.get_refresh_token(code)
        print(f"Access token: {oauth.access_token}")
        print(f"Refresh token: {oauth.refresh_token}")
        # Dont forget to store the refresh token so you can use it again later

asyncio.run(main())

Fleet API for Vehicles

The TeslaFleetApi class provides methods to interact with the Fleet API for vehicles. Here's a basic example:

import asyncio
import aiohttp
from tesla_fleet_api import TeslaFleetApi
from tesla_fleet_api.exceptions import TeslaFleetError

async def main():
    async with aiohttp.ClientSession() as session:
        api = TeslaFleetApi(
            access_token="<access_token>",
            session=session,
            region="na",
        )

        try:
            data = await api.vehicles.list()
            print(data)
        except TeslaFleetError as e:
            print(e)

asyncio.run(main())

For more detailed examples, see Fleet API for Vehicles.

Fleet API for Energy Sites

The EnergySites class provides methods to interact with the Fleet API for energy sites. Here's a basic example:

import asyncio
import aiohttp
from tesla_fleet_api import TeslaFleetApi
from tesla_fleet_api.exceptions import TeslaFleetError

async def main():
    async with aiohttp.ClientSession() as session:
        api = TeslaFleetApi(
            access_token="<access_token>",
            session=session,
            region="na",
        )

        try:
            energy_sites = await api.energySites.list()
            print(energy_sites)
        except TeslaFleetError as e:
            print(e)

asyncio.run(main())

For more detailed examples, see Fleet API for Energy Sites.

To pair an energy gateway's RSA key over the cloud and compose the resulting signed local LAN control (via the sibling aiopowerwall library) with a cloud fallback through EnergySiteRouter, see Energy: Local Control.

Fleet API with Signed Vehicle Commands

The VehicleSigned class provides methods to interact with the Fleet API using signed vehicle commands. Here's a basic example:

import asyncio
import aiohttp
from tesla_fleet_api import TeslaFleetApi
from tesla_fleet_api.tesla.vehicle.signed import VehicleSigned
from tesla_fleet_api.exceptions import TeslaFleetError

async def main():
    async with aiohttp.ClientSession() as session:
        api = TeslaFleetApi(
            access_token="<access_token>",
            session=session,
            region="na",
        )

        try:
            vehicle = VehicleSigned(api, "<vin>")
            data = await vehicle.wake_up()
            print(data)
        except TeslaFleetError as e:
            print(e)

asyncio.run(main())

For more detailed examples, see Fleet API with Signed Vehicle Commands.

Bluetooth for Vehicles

The TeslaBluetooth class provides methods to interact with Tesla vehicles using Bluetooth. Here's a basic example:

import asyncio
from bleak import BleakScanner
from tesla_fleet_api import TeslaBluetooth

async def main():
    scanner = BleakScanner()
    devices = await scanner.discover()
    for device in devices:
        if TeslaBluetooth().valid_name(device.name):
            print(f"Found Tesla vehicle: {device.name}")

asyncio.run(main())

For more detailed examples, see Bluetooth for Vehicles.

get_private_key(path) loads an existing EC private key or creates a new unencrypted PEM key file. Newly created key files are created owner-readable and owner-writable only (0600) from the start, with no write-then-chmod window, and concurrent creators fall back to reading the file that won the create race.

VehicleBluetooth keeps a held BLE connection alive during idle periods by default with a passive GATT read about every 20 seconds. Pass keepalive_interval=None (or 0) when creating the vehicle to disable it; leaving it enabled can keep an already-awake car awake longer, so disconnect or disable keepalive when vehicle sleep is preferred.

BLE connect/notify failures, and GATT writes rejected before backend I/O, raise BluetoothTransportError, a TeslaFleetError subclass, with the original transport exception chained as __cause__ when available. A GATT write that entered backend I/O and then failed or timed out is delivery-ambiguous and raises BluetoothTimeout/BluetoothUnconfirmedCommand instead. Mutating BLE commands use a confirmation ladder controlled by confirmation ("ack" by default) and raise_unconfirmed (False by default): an inconclusive lost acknowledgement resolves as best-effort success unless you opt in to BluetoothUnconfirmedCommand, while a command proven not to have applied raises BluetoothCommandFailed. See Bluetooth for Vehicles for the full ladder. Catch TeslaFleetError to handle Bluetooth transport failures (including bleak.exc.BleakError and builtin TimeoutError from ESPHome proxies) and response-wait timeouts through the same library error hierarchy.

VehicleBluetooth can also register persistent BLE broadcast listeners for unsolicited VCSEC VehicleStatus updates. Use typed listen_* helpers for the decoded vehicle-status fields, or listen_broadcast(domain, callback) for raw per-domain broadcast messages.

Routing and Failover

The Router class composes an ordered list of two-or-more backends that share a common method surface and dispatches each method call down the chain, automatically failing over on most errors. VehicleRouter and EnergySiteRouter are thin entity-specific subclasses. A common setup is a local VehicleBluetooth primary with a cloud fallback (e.g. a TeslemetryVehicle), so commands go over Bluetooth when the vehicle is reachable and route to the cloud otherwise:

import asyncio
import aiohttp
from tesla_fleet_api import TeslaBluetooth, Teslemetry
from tesla_fleet_api.router import VehicleRouter
from tesla_fleet_api.exceptions import TeslaFleetError

async def main():
    async with aiohttp.ClientSession() as session:
        # Primary: local Bluetooth
        tesla_bluetooth = TeslaBluetooth()
        await tesla_bluetooth.get_private_key("path/to/private_key.pem")
        primary = tesla_bluetooth.vehicles.create("<vin>", confirmation="verify")

        # Secondary (fallback): Teslemetry cloud
        teslemetry = Teslemetry(access_token="<access_token>", session=session)
        secondary = teslemetry.vehicles.create("<vin>")

        vehicle = VehicleRouter(primary, secondary)

        try:
            await vehicle.wake_up()
        except TeslaFleetError as e:
            print(e)

asyncio.run(main())

The constructor is Router(primary, secondary, *more_backends, health=None); the two-argument form shown above is fully backward compatible, and any number of extra backends may follow to extend the chain. Each call is tried on the first backend that has the method and, on any exception except BluetoothUnconfirmedCommand, retried on the next backend that has it, returning the first success (raising the last error only if every applicable backend fails). Non-callable attributes (e.g. vin) resolve to the first backend that has them.

By default the router attempts the primary and fails over on any error, with no up-front probe. You can also pass an explicit health check — a bool, a sync callable, or an async callable returning bool — to decide up front whether to route to the primary or skip straight to the rest of the chain. The health check gates only the primary (the first backend); later backends are reached purely through per-command failover.

EnergySiteRouter follows the same pattern for energy sites, pairing a duck-typed local EnergySite-shaped object (e.g. aiopowerwall's PowerwallEnergySite, no dependency added) with a cloud TeslemetryEnergySite fallback:

from tesla_fleet_api.router import EnergySiteRouter

router = EnergySiteRouter(local_energysite, teslemetry_energysite)
await router.operation(...)  # local first, cloud on failure

Router, VehicleRouter, and EnergySiteRouter are all importable from tesla_fleet_api.router (and, for backward compatibility, from tesla_fleet_api.tesla).

Enable DEBUG logging for tesla_fleet_api to see which backend served a routed call and why failover happened.

Warning: Because a failed call is replayed on the next backend, a non-idempotent command (e.g. honk_horn, actuate_trunk, door_unlock, charge_start) that fails mid-flight — after a backend may have already partially applied it — can be double-executed (or executed more than once across a longer chain) when it is retried on the next backend. This is a deliberate tradeoff of per-command failover. BluetoothUnconfirmedCommand is the exception: it propagates without failover because the BLE command may already have executed. When the primary is VehicleBluetooth, pass confirmation="verify" to resolve supported mutating command timeouts by state before they reach the router, and set raise_unconfirmed=True when callers must see still-ambiguous outcomes instead of the default best-effort success; callers needing exactly-once semantics for other commands should gate dispatch with an explicit health check or call the underlying backends directly.

Dispatch is implemented via __getattr__, which does not proxy dunder methods, so async with Router(...) does not manage a backend's BLE connection lifecycle (__aenter__/__aexit__). Commands still auto-connect on send; for explicit connect/disconnect reach through router.primary (or router.backends).

Debug Logging

Enable the tesla_fleet_api logger at DEBUG to see each command's command name, transport/backend, and result. In standalone scripts, configure a handler first:

import logging

logging.basicConfig(level=logging.DEBUG)
logging.getLogger("tesla_fleet_api").setLevel(logging.DEBUG)

Command log lines use transport=bluetooth, fleet, teslemetry, or tessie. Routers also emit backend=<ClassName> lines for each backend tried. See Bluetooth for Vehicles for examples and the signed-command naming details. REST responses that are valid JSON but not objects, such as null, lists, or scalars, are returned unchanged and log as result=success.

Teslemetry

The Teslemetry class provides methods to interact with the Teslemetry service. Here's a basic example:

import asyncio
import aiohttp
from tesla_fleet_api import Teslemetry
from tesla_fleet_api.exceptions import TeslaFleetError

async def main():
    async with aiohttp.ClientSession() as session:
        api = Teslemetry(
            access_token="<access_token>",
            session=session,
        )

        try:
            data = await api.vehicles.list()
            print(data)
        except TeslaFleetError as e:
            print(e)

asyncio.run(main())

For more detailed examples, see Teslemetry.

Tessie

The Tessie class provides methods to interact with the Tessie service. Here's a basic example:

import asyncio
import aiohttp
from tesla_fleet_api import Tessie
from tesla_fleet_api.exceptions import TeslaFleetError

async def main():
    async with aiohttp.ClientSession() as session:
        api = Tessie(
            access_token="<access_token>",
            session=session,
        )

        try:
            data = await api.vehicles.list()
            print(data)
        except TeslaFleetError as e:
            print(e)

asyncio.run(main())

For more detailed examples, see Tessie.