Python SDK

wallecpms is the official Python SDK for the Pipelet Headless CPMS API. It’s a thin, typed wrapper over the REST endpoints with sensible exceptions and a context-manager pattern for connection lifecycle.

Installation

From PyPI

pip install wallecpms

From source

git clone https://github.com/pipelet/cpms-headless.git
pip install ./cpms-headless/sdk

From a wheel

pip install wallecpms-1.0.0-py3-none-any.whl

Quick start

from wallecpms import HeadlessCPMS

cpms = HeadlessCPMS("http://localhost:8080", api_key="hcpms_live_...")

# List connected stations
for station in cpms.stations.list():
    print(station["station_id"], station.get("status"))

# Start a charging session
result = cpms.transactions.start("WALLBOX_001", connector_id=1, id_tag="RFID_123")
print(result)  # { "status": "Accepted", "transactionId": 42 }

# Stop charging by transaction ID
cpms.transactions.stop(transaction_id=42)

# Health check
print(cpms.system.health())

Context manager

For long-lived clients, use the context manager so the underlying HTTP session closes cleanly:

with HeadlessCPMS("http://localhost:8080", api_key="...") as cpms:
    print(cpms.stations.list())

Resource catalog

Resource	Methods
cpms.stations	list(), list_compact(), liveness(), reset(), disconnect(), change_availability(), unlock_connector()
cpms.transactions	list(), start(), stop(), stop_connector()
cpms.configuration	get(), set(), get_local_auth(), set_local_auth(), get_local_list_version(), send_local_list()
cpms.triggers	boot(), meter_values(), status(), status_round(), message(), datatransfer()
cpms.firmware	update(), diagnostics()
cpms.webhooks	list(), create(), get(), update(), delete(), events(), test()
cpms.hubject	remote_start(), remote_stop(), reserve(), sync()
cpms.system	health(), metrics_server(), metrics_websockets(), metrics_db_pool()

Error handling

The SDK raises typed exceptions on non-2xx responses:

from wallecpms import HeadlessCPMS, AuthenticationError, NotFoundError, CPMSError

cpms = HeadlessCPMS("http://localhost:8080", api_key="wrong-key")

try:
    cpms.stations.list()
except AuthenticationError:
    print("Invalid API key!")
except NotFoundError:
    print("Resource not found")
except CPMSError as e:
    print(f"API error: {e} (HTTP {e.status_code})")

Webhook management

# Create a webhook
wh = cpms.webhooks.create(
    name="My Integration",
    url="https://my-app.example.com/webhooks/ocpp",
    events=["StartTransaction", "StopTransaction", "StatusNotification"],
    secret_header="X-Webhook-Secret",
    secret_value="my-secret",
)

# List available event types
print(cpms.webhooks.events())

# Test a webhook
cpms.webhooks.test(wh["id"])

# Delete it
cpms.webhooks.delete(wh["id"])

Async support

wallecpms is currently sync-only. For high-concurrency workloads or async frameworks (FastAPI, Starlette), use the REST API directly with httpx.AsyncClient until the async port lands in v2.

Roadmap

A TypeScript SDK (@pipelet/cpms) is planned for Phase 3 of this developer portal — auto-generated from the same OpenAPI spec the REST gateway publishes.

Configuration patterns

import os
from wallecpms import HeadlessCPMS

cpms = HeadlessCPMS(
    base_url=os.environ["CPMS_URL"],
    api_key=os.environ["CPMS_API_KEY"],
    timeout=10.0,        # seconds, default 30
    verify_ssl=True,     # default True; set False for self-signed certs
)

Next

Onboarding workflow tutorial A complete example using the SDK to onboard a fleet.

Headless CPMS API reference Every endpoint the SDK wraps, with full parameter and response schemas.