tesla-fleet-api 1.6.4

Tesla Fleet API library for Python

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.

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.

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 and GATT write failures from VehicleBluetooth raise BluetoothTransportError, a TeslaFleetError subclass, with the original transport exception chained as __cause__. Catch TeslaFleetError to handle Bluetooth transport failures (including bleak.exc.BleakError and builtin TimeoutError from ESPHome proxies) and BluetoothTimeout response-wait timeouts through the same library error hierarchy.

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 error. 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>", verify_commands=True)

        # 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, 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.set_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. When the primary is VehicleBluetooth, pass verify_commands=True to resolve supported mutating command timeouts by state before failover; 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.

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.