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()
        print(relations)


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:
        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 (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.

EPEX prices (no login required)

The EPEX day-ahead market endpoint is public. Fetch it from an unauthenticated client, with no tokens and no login:

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(...):

Token state

Authentication

Persist every rotated pair via on_token_refresh. See Token rotation.

Data getters

Exceptions

All exceptions derive from EngieBeError, which carries an optional HTTP status:

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.

Development

This project uses uv and targets Python 3.14+.

uv sync
uv run python -m scripts.check

License

MIT, see LICENSE.