OpenADR 3 companion client with VEN/BL client framework, lifecycle management, and optional MQTT and webhook notification channels.
Built on top of openadr3 (Pydantic models, httpx HTTP client).
pip install python-oa3-client # core: VEN/BL clients, API access
pip install python-oa3-client[mqtt] # + MQTT notifications
pip install python-oa3-client[webhooks] # + webhook receiver
pip install python-oa3-client[mdns] # + mDNS/DNS-SD VTN discovery
pip install python-oa3-client[all] # everythingThe core package depends only on openadr3. Notification channels are optional extras:
| Extra | Adds | Dependency |
|---|---|---|
mqtt | MQTT broker connection, topic discovery, message collection | ebus-mqtt-client (paho-mqtt v2) |
webhooks | HTTP webhook receiver for VTN callbacks | Flask |
mdns | mDNS/DNS-SD VTN discovery (_openadr3._tcp.) | zeroconf |
all | All of the above | — |
BaseClient — auth, lifecycle, __getattr__ delegation to OpenADRClient
├── VenClient — VEN registration, program lookup, notification subscribe
└── BlClient — thin wrapper, client_type="bl", no VEN concepts
All OpenADRClient methods (raw HTTP, coerced entities, introspection) are available directly on both client types via __getattr__ delegation — no explicit delegation methods needed.
Two auth modes:
Direct token — provide a Bearer token directly:
ven=VenClient(url=vtn_url, token=my_token)OAuth2 client credentials — token fetched automatically on start():
ven=VenClient(
url=vtn_url,
client_id="my_client",
client_secret="my_secret",
)For the OpenADR 3 VTN Reference Implementation, the default auth uses basic credentials encoded as base64(client_id:secret):
importbase64bl_token=base64.b64encode(b"bl_client:1001").decode()
ven_token=base64.b64encode(b"ven_client:999").decode()Clients send a composed User-Agent header for server-side log identification:
python-oa3-client/0.4.0 openadr3/0.3.0 (node=a1b2c3d4e5f6)
Add your own identifier with the user_agent parameter:
ven=VenClient(
url=vtn_url,
token=token,
user_agent="my-thermostat/1.0 (contact@example.com)",
)
# UA: python-oa3-client/0.4.0 openadr3/0.3.0 (node=...) my-thermostat/1.0 (contact@example.com)Requires: pip install python-oa3-client[mdns]
The OpenADR 3.1.0 spec defines mDNS service discovery for local VTNs using service type _openadr3._tcp.. Clients can discover VTNs on the local network without a configured URL.
| Mode | Behavior |
|---|---|
"never" (default) | Skip mDNS, use configured url. Current behavior. |
"prefer_local" | Try mDNS; use discovered VTN if found; fall back to url; raise if neither. |
"local_with_fallback" | Try mDNS; fall back to configured url if not found (requires url). |
"require_local" | Try mDNS; raise if no VTN found. No url needed. |
fromopenadr3_clientimportVenClient# No URL needed — discovers VTN on the local networkwithVenClient(token=token, discovery="require_local") asven:
ven.register("my-thermostat")
events=ven.events()withVenClient(
url="https://cloud-vtn.example.com/openadr3/3.1.0",
token=token,
discovery="local_with_fallback",
discovery_timeout=3.0,
) asven:
# Uses local VTN if found, otherwise cloud URLven.register("my-thermostat")fromopenadr3_clientimportdiscover_vtnsvtns=discover_vtns(timeout=3.0)
forvinvtns:
print(f"{v.name} at {v.url} (version={v.version})")For testing mDNS discovery without modifying the VTN itself:
fromopenadr3_clientimportadvertise_vtnwithadvertise_vtn(
host="127.0.0.1",
port=8080,
base_path="/openadr3/3.1.0",
local_url="http://127.0.0.1:8080/openadr3/3.1.0",
version="3.1.0",
) asadv:
# VTN is now visible via mDNS# ... run discovery tests ...# Service unregistered on exitVenClient is the primary interface for VEN developers:
fromopenadr3_clientimportVenClientwithVenClient(url="http://vtn:8080/openadr3/3.1.0", token=token) asven:
# Register VEN (idempotent — finds existing or creates new)ven.register("my-thermostat-ven")
# Find a specific programpricing=ven.find_program_by_name("residential-pricing")
# Check notification supportifven.vtn_supports_mqtt():
mqtt=ven.add_mqtt("mqtts://broker:8883")
mqtt.start()
ven.subscribe(
program_names=["residential-pricing"],
objects=["EVENT"],
operations=["CREATE", "UPDATE"],
channel=mqtt,
)
msgs=mqtt.await_messages(1, timeout=30.0)
else:
events=ven.poll_events(program_name="residential-pricing")
# All OpenADRClient methods work via __getattr__resp=ven.get_subscriptions()
reports=ven.reports()ven.register("my-ven")
print(ven.ven_id) # "ven-abc-123"print(ven.ven_name) # "my-ven"# Query by name (caches ID)program=ven.find_program_by_name("residential-pricing")
# Cached name→ID resolutionpid=ven.resolve_program_id("residential-pricing")notifiers=ven.discover_notifiers()
supports_mqtt=ven.vtn_supports_mqtt()
# Extract broker URIs from /notifiers — handles both spec# ({"MQTT": {"URIS": [...]}}) and VTN-RI ([{"transport": "MQTT", "url": ...}])# response shapes. Returns [] if MQTT isn't advertised.uris=ven.get_mqtt_broker_uris()
ifuris:
mqtt=ven.add_mqtt(uris[0])URIs from /notifiers may use any of the mqtt://, mqtts://, tcp://, or
ssl:// schemes (or no scheme at all — broker.example.com:1883).
MqttChannel and MQTTConnection accept all of these via
normalize_broker_uri.
Default to the registered ven_id when called without arguments:
ven.register("my-ven")
resp=ven.get_mqtt_topics_ven() # uses registered ven_idresp=ven.get_mqtt_topics_ven_events()
resp=ven.get_mqtt_topics_ven("other-id") # explicit ven_idFor business logic (creating programs, events):
fromopenadr3_clientimportBlClientwithBlClient(url=vtn_url, token=bl_token) asbl:
bl.create_program({
"programName": "tariff-program",
"programType": "PRICING_TARIFF",
"country": "US",
"principalSubdivision": "CA",
"intervalPeriod": {"start": "2024-01-01T00:00:00Z", "duration": "P1Y"},
})
bl.create_event({...})Requires: pip install python-oa3-client[mqtt]
mqtt=ven.add_mqtt("mqtt://broker:1883", client_id="my-ven-mqtt")
mqtt.start()
# Manual topic subscriptionmqtt.subscribe_topics(["openadr3/#"])
# Or use ven.subscribe() for program-aware subscriptionven.subscribe(
program_names=["residential-pricing"],
objects=["EVENT"],
operations=["CREATE", "UPDATE"],
channel=mqtt,
)
msgs=mqtt.await_messages(n=1, timeout=10.0)
formsginmsgs:
print(msg.topic, msg.payload)
mqtt.stop()TLS connections: use mqtts:// scheme (default port 8883).
Requires: pip install python-oa3-client[webhooks]
webhook=ven.add_webhook(
port=0, # OS-assigned ephemeral portbearer_token="my-secret",
callback_host="192.168.1.50", # IP reachable from VTN
)
webhook.start()
print(webhook.callback_url) # "http://192.168.1.50:54321/notifications"# Subscribe creates VTN subscription with callback URLven.subscribe(
program_names=["residential-pricing"],
objects=["EVENT"],
operations=["CREATE", "UPDATE"],
channel=webhook,
)
msgs=webhook.await_messages(n=1, timeout=10.0)
webhook.stop()Channels are created but not started automatically. You control the lifecycle:
mqtt=ven.add_mqtt(broker_url) # Created, not connectedmqtt.start() # Connected# ... use ...mqtt.stop() # DisconnectedWhen VenClient stops (via stop() or context manager exit), all channels are stopped automatically.
MQTTMessage:
| Field | Type | Description |
|---|---|---|
topic | str | MQTT topic |
payload | Any | Parsed JSON, or coerced Notification |
time | float | Unix timestamp |
raw_payload | bytes | Original bytes |
WebhookMessage:
| Field | Type | Description |
|---|---|---|
path | str | URL path |
payload | Any | Parsed JSON, or coerced Notification |
time | float | Unix timestamp |
raw_payload | bytes | Original request body |
python-oa3-client does no datetime parsing of its own — all time handling is
delegated to openadr3,
which is the reference compliant implementation of the cross-implementation
zone-handling rule.
- Datetime fields on coerced entities (e.g.
event.created,interval_period.start) and on coerced notification payloads (delivered byMqttChannelandWebhookChannel) are surfaced as zone-awarependulum.DateTime. - The wire string's UTC offset is the source of truth and is preserved
end-to-end.
Z,+00:00,-07:00,+05:30round-trip exactly — no normalization. See python-oa3 README — Time and Timezones for the canonical specification and round-trip table. - Cross-implementation parity with
clj-oa3-client(which surfacesjava.time.ZonedDateTimeunder the same rule).
Propagation through this client's MQTT and webhook parse paths is locked in
by tests/test_time_propagation.py.
All OpenADRClient methods are available on both VenClient and BlClient via __getattr__:
# Raw HTTP (returns httpx.Response)resp=ven.get_programs(skip=0, limit=10)
resp=ven.create_subscription({...})
# Coerced entities (returns Pydantic models)programs=ven.programs()
event=ven.event("evt-001")
reports=ven.reports()
subscriptions=ven.subscriptions()
# Introspection (requires spec_path)routes=ven.all_routes()
scopes=ven.endpoint_scopes("/programs", "get")The standalone MQTTConnection, WebhookReceiver, extract_topics, normalize_broker_uri, and detect_lan_ip are still exported for direct use.
examples/smoke_test.py— integration test against live VTN-RI and Mosquittoexamples/smoke_test_mdns.py— mDNS discovery integration test (advertise + discover + connect)examples/ven_workflow.py— documented VEN developer workflowexamples/ven_mdns.py— zero-config VEN that discovers its VTN via mDNSdoc/ven-bl-client-guide.md— VEN & BL client use-case walkthrough
git clone https://github.com/grid-coordination/python-oa3-client
cd python-oa3-client
pip install -e ".[dev]"
pytest tests/ -vIssues, Discussions, and pull requests are welcome — see CONTRIBUTING.md for the workflow (and the dev commands: tests, lint, format, pre-commit). In short:
- Questions, API/design discussion, VTN behavior gaps → Discussions
- Confirmed bugs, channel/discovery fixes, doc errors → Issues
- Patches → pull requests; please open a Discussion or Issue first for non-trivial changes (new channel types, new discovery modes, new auth modes, new lifecycle hooks)
Bugs in raw HTTP, coerced entities, or spec-level shapes likely belong in python-oa3 rather than here.
MIT License — Copyright (c) 2026 Clark Communications Corporation