Skip to navigation

Python SDK

Typed sync and async clients for Python 3.9+. Package ocpplab on PyPI.

Install

pip install ocpplab

Create a client

import os
from ocpplab import OcpplabSDK
client = OcpplabSDK(token=os.environ["OCPPLAB_TOKEN"])
ArgumentDefaultUse it to
token—Authenticate. See Authentication.
base_urlhttps://api.ocpplab.ioCall another host, such as a local API.
timeout60Fail a request that takes longer, in seconds.
httpx_client—Bring your own httpx.Client, for example to set a proxy.

Per request, pass request_options={"timeout_in_seconds": 10, "max_retries": 1}.

Call the API

Methods follow the API’s domains and are snake_case, like the request and response fields.

for brand in client.evcharger.catalog.list_brands():
    print(brand.slug, brand.name)
status = client.evcharger.commands.get_status("cmd-123")
print(status.status)

Every method, its parameters and a runnable snippet are in the API Reference: pick Python in any code example.

Async client

AsyncOcpplabSDK has the same methods; await them.

import asyncio
import os
from ocpplab import AsyncOcpplabSDK
async def main() -> None:
    client = AsyncOcpplabSDK(token=os.environ["OCPPLAB_TOKEN"])
    status = await client.evcharger.commands.get_status("cmd-123")
    print(status.status)
asyncio.run(main())

Pagination

List methods return a pager. Iterate it to walk every item, or go page by page:

pager = client.evcharger.catalog.list_brands(limit=20)
for page in pager.iter_pages():
    for brand in page:
        print(brand.slug)

Types

Request and response models are Pydantic models under ocpplab.types:

from ocpplab.types import CommandTarget
target = CommandTarget(identity="DEMO-CP-001")

Full example

quickstart.py
"""Put virtual chargers online against your CSMS, then wait until they connect.
    export OCPPLAB_TOKEN="<api-key>"
    export CSMS_WS_URL="wss://csms.example.com/ocpp"
    uv run python examples/quickstart.py
Creating a location returns 202 at once; the chargers deploy in the background,
so this polls their deployment stage until every one is online.
"""
from __future__ import annotations
import os
import time
from ocpplab import OcpplabSDK
from ocpplab.types import VirtualLocationChargerItem, VirtualLocationCoordinates
IDENTITIES = ["DEMO-CP-001", "DEMO-CP-002"]
def main() -> None:
    # Defaults to https://api.ocpplab.io; pass base_url= for another host.
    client = OcpplabSDK(token=os.environ["OCPPLAB_TOKEN"])
    accepted = client.evcharger.location.create(
        name="Demo hub",
        address="1 Parvis de la Defense",
        city="Puteaux",
        country="FRA",
        coordinates=VirtualLocationCoordinates(lat=48.892, lng=2.238),
        chargers=[
            VirtualLocationChargerItem(
                identity=identity,
                brand_slug="wallbox",
                model="wallbox-pulsar-plus",
                connector_type="Type2",
                ws_url=os.environ["CSMS_WS_URL"],
            )
            for identity in IDENTITIES
        ],
    )
    print(f"location {accepted.location_id}: {accepted.detail}")
    deadline = time.monotonic() + 600
    while True:
        chargers = client.evcharger.deployment_runtime_ops.status(
            identity=IDENTITIES
        ).items
        for charger in chargers:
            print(charger.identity, charger.stage, charger.outcome, charger.hint or "")
        if chargers and all(charger.stage == "online" for charger in chargers):
            return
        if any(charger.outcome in ("failed", "stuck") for charger in chargers):
            raise SystemExit("a charger failed to deploy; see the hint above")
        if time.monotonic() > deadline:
            raise SystemExit("chargers were not online after 10 minutes")
        time.sleep(5)
if __name__ == "__main__":
    main()

Next