This document provides detailed examples for using the Fleet API for energy sites.
The TeslaFleetApi class provides methods to interact with the Fleet API for energy sites. First, create an EnergySite instance using the energy site ID (which can be found via api.products()):
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_site = api.energySites.create(12345)
data = await energy_site.site_info()
print(data)
except TeslaFleetError as e:
print(e)
asyncio.run(main())Enable the tesla_fleet_api logger at DEBUG to log each Fleet API request's
final path segment, transport=fleet, and result:
import logging
logging.basicConfig(level=logging.DEBUG)
logging.getLogger("tesla_fleet_api").setLevel(logging.DEBUG)Responses that are valid JSON but not objects, such as null, lists, or
scalars, are returned unchanged and log as result=success.
You can adjust the backup reserve for a specific energy site using its ID:
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_site = api.energySites.create(12345)
backup_reserve_response = await energy_site.backup(backup_reserve_percent=20)
print(backup_reserve_response)
except TeslaFleetError as e:
print(e)
asyncio.run(main())You can get the backup (off-grid) event history of a specific energy site using its ID:
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_site = api.energySites.create(12345)
backup_history_response = await energy_site.backup_history(period="day")
print(backup_history_response)
except TeslaFleetError as e:
print(e)
asyncio.run(main())You can get the charging history of a specific energy site using its ID:
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_site = api.energySites.create(12345)
charge_history_response = await energy_site.charge_history(start_date="2022-01-01", end_date="2022-01-31")
print(charge_history_response)
except TeslaFleetError as e:
print(e)
asyncio.run(main())You can get the energy measurements of a specific energy site using its ID:
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_site = api.energySites.create(12345)
energy_history_response = await energy_site.energy_history(period="day")
print(energy_history_response)
except TeslaFleetError as e:
print(e)
asyncio.run(main())You can allow or disallow charging from the grid and exporting energy to the grid for a specific energy site using its ID:
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_site = api.energySites.create(12345)
grid_import_export_response = await energy_site.grid_import_export(disallow_charge_from_grid_with_solar_installed=True)
print(grid_import_export_response)
except TeslaFleetError as e:
print(e)
asyncio.run(main())You can get the live status of a specific energy site using its ID:
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_site = api.energySites.create(12345)
live_status_response = await energy_site.live_status()
print(live_status_response)
except TeslaFleetError as e:
print(e)
asyncio.run(main())You can adjust the off-grid vehicle charging backup reserve for a specific energy site using its ID:
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_site = api.energySites.create(12345)
off_grid_vehicle_charging_reserve_response = await energy_site.off_grid_vehicle_charging_reserve(off_grid_vehicle_charging_reserve_percent=10)
print(off_grid_vehicle_charging_reserve_response)
except TeslaFleetError as e:
print(e)
asyncio.run(main())You can set the operation mode for a specific energy site using its ID:
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_site = api.energySites.create(12345)
operation_mode_response = await energy_site.operation(default_real_mode="self_consumption")
print(operation_mode_response)
except TeslaFleetError as e:
print(e)
asyncio.run(main())You can get information about a specific energy site using its ID:
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_site = api.energySites.create(12345)
site_info_response = await energy_site.site_info()
print(site_info_response)
except TeslaFleetError as e:
print(e)
asyncio.run(main())You can update the storm watch participation for a specific energy site using its ID:
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_site = api.energySites.create(12345)
storm_mode_response = await energy_site.storm_mode(enabled=True)
print(storm_mode_response)
except TeslaFleetError as e:
print(e)
asyncio.run(main())You can update the time of use settings for a specific energy site using its ID:
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_site = api.energySites.create(12345)
time_of_use_settings_response = await energy_site.time_of_use_settings(settings={})
print(time_of_use_settings_response)
except TeslaFleetError as e:
print(e)
asyncio.run(main())The top-level get_tariff_periods helper resolves the current buy and sell
rates from a raw tariff_content_v2 object without making any API calls.
unwrap_tariff_v2 accepts the site_info() response envelope, the
tou_settings write envelope, or a bare tariff object:
from datetime import datetime
from zoneinfo import ZoneInfo
from tesla_fleet_api import get_tariff_periods, unwrap_tariff_v2
site_info = await energy_site.site_info()
tariff = unwrap_tariff_v2(site_info)
site_timezone = ZoneInfo("<installation_time_zone from site_info>")
now = datetime.now(site_timezone)
resolution = get_tariff_periods(tariff, now, horizon_hours=24)now must be timezone-aware and expressed in the site's local timezone because
the tariff object does not carry its own timezone. A naive now raises
ValueError. The result contains the current buy and sell rates, the current
period's start, the next change, the currency, and (when horizon_hours is
provided) a contiguous timeline of upcoming periods. Gaps in the buy schedule
are preserved as one merged TariffPeriod with both rates set to None; the
gap ends when the tariff resumes, or at the horizon deadline if it does not.
The helper returns None when no tariff season covers now. Missing rates
remain None, while a real zero price remains 0.0.
Null or malformed inputs passed to unwrap_tariff_v2 raise InvalidResponse.
This includes every accepted envelope or bare tariff object whose extracted
tariff lacks either seasons or energy_charges.
Energy gateways (Powerwalls, etc.) support gRPC commands sent via POST /api/1/energy_sites/{id}/command. These are undocumented Tesla API endpoints that communicate directly with the gateway hardware. All device command methods require the energy_cmds scope.
These commands are unsigned and cloud-only. The actuating set_island_mode/go_off_grid/reconnect_grid methods on EnergySite raise SignedCommandRequired unconditionally - the gateway can accept that unsigned command without physically taking effect, so there is no cloud-only way to actuate it. For registering a key and pairing it with a signed local LAN control path via the sibling aiopowerwall library, see Energy: Local Control.
Use list_authorized_clients() only as a secondary, best-effort cloud check
while pairing a local key. The reliable verification is a successful signed
local read through the paired LAN client; the cloud list endpoint can return
an empty or null response even when it is not useful for proving local key
readiness.
get_rsa_private_key(path) loads an existing RSA private key for gateway
client registration 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.
| Method | Category | Description |
|---|---|---|
get_system_info() |
Common | Firmware version, device type, part number, serial number, DIN |
get_networking_status() |
Common | WiFi, Ethernet, and cellular connectivity status |
wifi_scan() |
Common | Scan for available WiFi networks |
get_device_cert() |
Common | Device certificate (subject, issuer, validity) |
list_authorized_clients() |
Authorization | Best-effort cloud listing of paired keys; not authoritative for local key verification |
add_authorized_client() |
Authorization | Register a public key for local signed LAN control |
remove_authorized_client() |
Authorization | Remove a paired public key; Fleet API route is inferred and not hardware-verified |
get_signed_commands_public_key() |
Authorization | Gateway's public key for signed commands |
get_backup_events() |
TEG | Backup event history (may timeout on some firmware) |
schedule_backup_event() |
TEG | Schedule a manual backup event |
cancel_backup_event() |
TEG | Cancel a scheduled backup event |
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_site = api.energySites.create(12345)
# Get gateway system information
system_info = await energy_site.get_system_info()
print(system_info)
# Get networking status
networking = await energy_site.get_networking_status()
print(networking)
# List authorized clients (paired keys)
clients = await energy_site.list_authorized_clients()
print(clients)
except TeslaFleetError as e:
print(e)
asyncio.run(main())