dmn-sdk 0.4.0

Python SDK for the DMN mesh — call tenant APIs from your code.

dmn-sdk

Python client for the DMN mesh. Call tenant APIs from your own code with one import.

Install

From a checkout:

pip install ./sdk/python/

From PyPI (v0.282.3+):

pip install dmn-sdk

Requires Python 3.9+. Pulls in requests and websocket-client.

Quickstart

import dmn_sdk as dmn

client = dmn.connect()                                 # reads DMN_API_TOKEN env
ingestor = client.tenant("/tata/dmn/acme").app("ingestor")

# Unary POST
resp = ingestor.post("/telemetry", json={"sensor": "s1", "value": 23.4})
print(resp.status_code, resp.json())

# Subscribe to a topic
for msg in ingestor.subscribe("telemetry"):
    print(msg.seq, msg.payload)

Authentication

Three sources, in priority order:

1. Explicit kwargs to dmn.connect(token=..., base_url=...)
2. Environment: DMN_API_TOKEN + DMN_API_BASE_URL + (optional) DMN_PROFILE
3. Config file ~/.dmn/credentials (INI-style)

Env var setup

# Mint a token at /ui/dev/tokens/ on the dashboard,
# then export:
export DMN_API_TOKEN="eyJhbGc...the wire JWT..."
export DMN_API_BASE_URL="https://console.dmn.inno8cube.com"

Config file setup

~/.dmn/credentials:

[default]
token = eyJhbGc...
base_url = https://console.dmn.inno8cube.com

[acme-prod]
token = eyJhbGc...
base_url = https://console.dmn.acme.io

Pick a profile:

client = dmn.connect(profile="acme-prod")

Or via env: export DMN_PROFILE=acme-prod.

API reference

dmn.connect(...)

def connect(
    token: str | None = None,
    base_url: str | None = None,
    *,
    profile: str | None = None,
    config_path: str | None = None,
    timeout: float = 30,
    ws_timeout: float = 30,
    verify_tls: bool = True,
    user_agent: str | None = None,
) -> Client

Returns a Client. Raises dmn.ConfigError if no source can supply token + base_url.

Client.tenant(name)

def tenant(self, name: str) -> Tenant

name must be a full NDN name starting with /, e.g. "/tata/dmn/acme". Returns a Tenant.

Tenant.app(name)

def app(self, name: str) -> App

Returns an App handle.

Unary methods on App

app.get(path, **kwargs)     -> Response
app.post(path, **kwargs)    -> Response
app.put(path, **kwargs)     -> Response
app.patch(path, **kwargs)   -> Response
app.delete(path, **kwargs)  -> Response

**kwargs are passed through to requests.Session.request — most useful ones: json=, data=, params=, headers=, timeout=.

App.subscribe(topic)

def subscribe(self, topic: str) -> Iterator[Message]

Opens a WebSocket to the gateway's subscribe bridge. Yields Message(seq, topic, payload, ts) objects as they arrive. The iterator runs until the server closes the connection or you break. Re-iterating opens a fresh connection.

Response

resp.status_code  -> int
resp.ok           -> bool          # True for 2xx
resp.headers      -> dict[str,str]
resp.text         -> str
resp.content      -> bytes
resp.json()       -> Any
resp.raw          -> requests.Response  # power-user escape hatch

Message

@dataclass(frozen=True)
class Message:
    seq: int          # per-topic monotonic sequence number
    topic: str        # which topic this message was on
    payload: Any      # decoded JSON body
    ts: int           # unix timestamp (seconds) at the publisher

Errors

All raised by the SDK descend from dmn.DMNError:

Exception	When
dmn.ConfigError	Missing token/base_url, malformed config file, unknown profile
dmn.AuthError	Gateway returned 401 or 403
dmn.NotFoundError	Gateway returned 404 (tenant/app not in routing table)
dmn.BackendError	5xx from the backend, transport failures, gateway-streamed error frame
dmn.APIError	Other non-2xx responses

Most application code wants:

try:
    resp = ingestor.post("/telemetry", json={...})
except dmn.AuthError:
    # Token issue — re-mint or re-auth
    raise
except dmn.NotFoundError:
    # Misconfigured route
    raise
except dmn.BackendError as e:
    # Retryable
    print(f"retrying: {e}")

Examples

See examples/:

• post_telemetry.py — single-shot POST to the reference ingestor
• subscribe_topic.py — WebSocket-backed iterator consuming a topic

Dev notes

# Run tests
cd sdk/python
pip install -e ".[dev]"
pytest -v

# Lint (if installed)
ruff check dmn_sdk tests

Versioning

The SDK version tracks the DMN agent's compatible wire format. v0.282.x is the first stable line; breaking changes to the gateway's REST/WS surface will bump the major. Backwards-compatible additions bump the minor.

License

Restricted-Use Evaluation Licence — see ../../docs/RESTRICTED-USE-LICENSE.md.