mxaccessgw/clients/python/PythonClientDesign.md

Python Client Detailed Design

Purpose

Provide an async Python client package for MXAccess Gateway, plus a test CLI and unit tests. The Python client should be useful for automation, diagnostics, and test harnesses.

Follow the Python Style Guide for handwritten code and the Protobuf Style Guide for generated contract inputs.

Package Layout

Recommended layout:

clients/python/
  pyproject.toml
  src/zb_mom_ww_mxgateway/
    __init__.py
    client.py
    session.py
    options.py
    auth.py
    values.py
    errors.py
    generated/
  src/zb_mom_ww_mxgateway_cli/
    __main__.py
    commands.py
  tests/

Expected dependencies:

• grpcio
• grpcio-tools
• protobuf
• click or typer
• pytest
• pytest-asyncio

Library API

Use async-first API. A sync wrapper can be added later if needed.

Suggested API:

client = await GatewayClient.connect(
    endpoint="localhost:5000",
    api_key=api_key,
    plaintext=True,
)

session = await client.open_session()
server = await session.register("python-client")
item = await session.add_item(server, "TestChildObject.TestInt")
await session.advise(server, item)

async for event in session.stream_events():
    ...

await session.close()
await client.close()

Classes:

class GatewayClient:
    @classmethod
    async def connect(cls, options: ClientOptions) -> "GatewayClient": ...
    async def open_session(self, options: OpenSessionOptions | None = None) -> "Session": ...
    async def invoke(self, request: MxCommandRequest) -> MxCommandReply: ...
    async def close(self) -> None: ...

class Session:
    session_id: str
    async def register(self, client_name: str) -> int: ...
    async def add_item(self, server_handle: int, item: str) -> int: ...
    async def add_item2(self, server_handle: int, item: str, context: str) -> int: ...
    async def advise(self, server_handle: int, item_handle: int) -> None: ...
    async def add_item_bulk(self, server_handle: int, tag_addresses: Sequence[str]) -> list[SubscribeResult]: ...
    async def advise_item_bulk(self, server_handle: int, item_handles: Sequence[int]) -> list[SubscribeResult]: ...
    async def remove_item_bulk(self, server_handle: int, item_handles: Sequence[int]) -> list[SubscribeResult]: ...
    async def unadvise_item_bulk(self, server_handle: int, item_handles: Sequence[int]) -> list[SubscribeResult]: ...
    async def subscribe_bulk(self, server_handle: int, tag_addresses: Sequence[str]) -> list[SubscribeResult]: ...
    async def unsubscribe_bulk(self, server_handle: int, item_handles: Sequence[int]) -> list[SubscribeResult]: ...
    async def write(self, server_handle: int, item_handle: int, value: MxValueInput, user_id: int = 0) -> None: ...
    async def stream_events(self) -> AsyncIterator[MxEvent]: ...
    async def close(self) -> None: ...

Authentication

Use gRPC metadata:

metadata = (("authorization", f"Bearer {api_key}"),)

Provide a metadata helper that all unary and streaming calls use. Redact API keys in exceptions and CLI output.

TLS

Support:

• insecure channel for local development,
• TLS channel with default roots,
• custom root certificate file.

Trust posture (trust-on-first-use)

The gateway can serve a self-signed certificate it generates itself (it has no PKI). grpc-python exposes no per-channel skip-verify hook, so the client cannot "accept any certificate" the way the other clients do. Instead, when the channel is not plaintext and neither ca_file nor require_certificate_validation is set, the TLS default is trust-on-first-use: the client fetches the server's presented certificate once via ssl.get_server_certificate (an unverified probe), pins it as the channel's only trust root, and — because the generated certificate always carries a localhost SAN — defaults grpc.ssl_target_name_override to localhost when no server_name_override was supplied (tolerating dial-by-IP or a hostname mismatch). A failed probe is surfaced as a transport error naming the endpoint.

To verify the gateway instead:

• set ca_file to verify against a specific CA, or
• set require_certificate_validation=True to verify against the system trust roots.

Both bypass the TOFU path.

Streaming

Expose stream_events as an async iterator. Canceling the task should cancel the gRPC stream.

Do not hide stream errors. Convert common auth/session errors into typed exceptions.

Error Handling

Define typed exceptions:

MxGatewayError
MxGatewayTransportError
MxGatewayAuthenticationError
MxGatewayAuthorizationError
MxGatewaySessionError
MxGatewayWorkerError
MxGatewayCommandError
MxAccessError

MxGatewayCommandError should include the raw protobuf reply when available.

Test CLI

Entry point:

mxgw-py

Recommended commands:

mxgw-py version
mxgw-py smoke --endpoint localhost:5000 --api-key-env MXGATEWAY_API_KEY --plaintext --item TestChildObject.TestInt
mxgw-py stream-events --session-id <id> --json
mxgw-py write --session-id <id> --server-handle 1 --item-handle 1 --type int32 --value 123

Use click or typer. JSON output should be stable for test automation.

Unit Tests

Use pytest and pytest-asyncio.

Use fake generated stubs or an in-process test gRPC server where practical.

Required tests:

• API key metadata injection,
• API key redaction,
• insecure and TLS channel option construction,
• request construction for method helpers,
• value conversion from Python values,
• status/error mapping,
• async event iteration,
• stream cancellation,
• CLI parsing,
• JSON output.

Integration Tests

Skip unless:

MXGATEWAY_INTEGRATION=1

Use bounded smoke flow and always attempt close_session in finally.

Packaging

Use pyproject.toml. Publishable package name should be stable, for example:

zb-mom-ww-mxaccess-gateway-client

Generated protobuf code should be regenerated through a documented command, not edited by hand.

Related Documentation

• Client Libraries Detailed Design
• Client Proto Generation
• Client Packaging
• Python Style Guide