aioengiebelgium

Async Python client for the ENGIE Belgium consumer API. Requires Python >= 3.14.

Unofficial and reverse-engineered: not endorsed by ENGIE, and it may break without notice whenever ENGIE changes their API.

Install

uv add aioengiebelgium

Documentation

The documentation is hosted at aioengiebelgium.readthedocs.io.

Usage

tokens.json in the example below is demo storage only. Restrict the file's permissions or use your application's secure storage for the token pair.

import asyncio
import json
from pathlib import Path

from aioengiebelgium import EngieBeClient, MfaMethod

_TOKEN_FILE = Path("tokens.json")


async def persist_tokens(access_token: str, refresh_token: str) -> None:
    # Save the token pair for the next refresh.
    payload = json.dumps({"access": access_token, "refresh": refresh_token})
    await asyncio.to_thread(_TOKEN_FILE.write_text, payload)


async def first_login() -> None:
    # First run: log in with MFA. on_token_refresh saves the tokens.
    async with EngieBeClient(on_token_refresh=persist_tokens) as client:
        flow = await client.async_start_authentication(
            "user@example.com",
            "password",
            mfa_method=MfaMethod.SMS,
        )
        # Once you receive your MFA code, then:
        await flow.async_submit_mfa("123456")

        relations = await client.async_get_customer_account_relations()
        for relation in relations.accounts:
            for agreement in relation.customer_account.business_agreements:
                print(agreement.business_agreement_number)


async def next_run() -> None:
    # Later runs: reuse the saved tokens and skip the login entirely.
    raw = await asyncio.to_thread(_TOKEN_FILE.read_text)
    tokens = json.loads(raw)

    async with EngieBeClient(
        access_token=tokens["access"],
        refresh_token=tokens["refresh"],
        on_token_refresh=persist_tokens,
    ) as client:
        # A business agreement number printed by first_login().
        contracts = await client.async_get_energy_contracts("1234567890")
        print(contracts)


if _TOKEN_FILE.exists():
    asyncio.run(next_run())
else:
    asyncio.run(first_login())

Token rotation

ENGIE rotates the refresh token on every refresh call, and the previous refresh token becomes invalid immediately. If you pass an on_token_refresh callback, it receives the new (access_token, refresh_token) pair whenever the tokens change: after a completed login, a manual refresh, an automatic refresh before expiry, or a refresh triggered by a 401. You must persist that pair durably. If you keep using the old refresh token after a rotation, the next refresh fails and the user has to log in again.

The client logs an exception raised by on_token_refresh and continues. It does not re-raise it, so handle and report storage failures inside the callback.

EPEX prices (no login required)

The EPEX day-ahead market endpoint is public. It works on a client with no tokens and no login. A logged-in client can call it too and sends its token.

import asyncio
from datetime import UTC, datetime, timedelta

from aioengiebelgium import EngieBeClient


async def main() -> None:
    async with EngieBeClient() as client:
        now = datetime.now(UTC)
        prices = await client.async_get_epex_prices(now, now + timedelta(days=1))
        print(prices)


asyncio.run(main())

API reference

Constructor

EngieBeClient(session=None, *, ...). Every argument after session is keyword-only.

Close the client with await client.close(), or use it as async with EngieBeClient(...) as client:. The session is closed only if the client created it. A closed client raises EngieBeClientClosedError.

Token state

Authentication

Persist every rotated pair via on_token_refresh. See Token rotation.

Data getters

Response models that hold lists carry skipped_entries, the number of malformed entries the parser dropped. The count includes entries dropped from nested lists, such as the business agreements of an account or the registers of a meter read. A meter read whose registers are all malformed is dropped and counts once. A non-zero count means the response is incomplete.

Service actions

These calls change the customer's ENGIE service. They were built from the ENGIE Smart App's API contract and have not been run against a live account.

Exceptions

Client errors derive from EngieBeError, which carries an optional HTTP status:

Catch a subclass before its parent. Invalid arguments raise ValueError before any request is sent:

• A business agreement or customer account number that is not all digits. Spaces are removed before the check.
• An EAN that is not digits with an optional _ID<n> suffix. A solar surplus delivery point id must carry the suffix.
• A month outside 1-12, a year outside 2000-2100 or a day outside the requested month.
• A start_date after the end_date, or an EPEX datetime without a timezone. The EPEX getter does not check the order of from_dt and to_dt.
• A vehicle_id or session_id that is not a positive integer.
• A negative page_number or a page_size below 1.
• For async_get_meter_reads: only one of the two dates, or latest=True combined with a date range.

EPEX day-ahead prices are published in the afternoon for the following day. Requesting a window whose prices have not been published yet raises EngieBeEpexNotPublishedError. Retry later.

Enums and helpers

The other exported enums (TouSlotCode, SolarSurplusLevel, SolarInferenceKey, VehicleChargeStatus, VehiclePolicyState, ChargingSessionType, ChargingSessionStatus, ChargingSessionSource, SmartChargeOutcomeState, ChargeSettingMode) list the known values of model fields. TouSlotCode covers the time-of-use schedules. Prices and usage details keep the raw uppercase slot codes.

bare_ean(ean) strips the delivery-point suffix (_ID1) from an EAN. ean_with_delivery_point_suffix(ean) appends it. __version__ holds the installed package version.

Development

This project uses uv and targets Python 3.14+.

uv sync
uv run python -m scripts.check

License

MIT, see LICENSE.