Docs / Contact / sdks/python/getting-started

Python SDK: Getting Started

Go from zero to publishing device data in under a minute. The SDK handles batching, optional offline storage, and connectivity checks, so you just write your application logic.

Requirements: Python 3.9+

Install#

The SDK installs straight from GitHub — with uv (recommended):

bash

uv add git+https://github.com/tendrl-inc-labs/contact-python

or with plain pip:

bash

pip install git+https://github.com/tendrl-inc-labs/contact-python

Either way the package is tendrl — the imports below work unchanged.

Need an API key?

Every client needs an entity API key. If you haven't created one yet, create your first entity and copy its API key from the Connection Instructions dialog.

Your First Message (4 Lines)#

python

from tendrl import Client

client = Client(api_key="your_api_key")
client.start()
client.publish({"temperature": 23.5, "humidity": 60}, tags=["sensor"])

That's it. The message is queued, batched, and delivered to Contact. Tags route it to your flows and connectors automatically.

Handling a Bad Key#

If no key is supplied (and the TENDRL_KEY environment variable is unset), the constructor raises immediately:

python

from tendrl.client import Client, APIException

try:
    client = Client(api_key=api_key)
except APIException as e:
    print(f"Client could not start: {e}")  # e.g. missing API key
    raise

A key that is present but invalid is only detected when the client first contacts Contact: the publish fails as an authentication error rather than raising here. publish() is asynchronous, so it cannot hand that failure back to you — the SDK reports it on the tendrl logger instead (see Knowing when messages don't arrive). Add Client(debug=True) during setup to also print each send attempt.

Pointing at a different server#

By default the client talks to https://app.tendrl.com. Pass app_url or set the TENDRL_APP_URL environment variable to point it somewhere else:

python

client = Client(api_key="your_api_key", app_url="http://localhost:8000")
bash

TENDRL_APP_URL=http://localhost:8000 python app.py

Either form accepts a bare origin (http://localhost:8000) or a full base URL already ending in /api. The same variable works for the Go and JavaScript SDKs and the Nano Agent — see Testing against a local or staging stack.

Automate with Tether#

Most IoT work is "read a sensor, publish the result, repeat." The @tether decorator does exactly that: wrap any function and its return value is automatically published.

python



@client.tether(tags=["sensor", "environment"])
def read_sensors():
    # Replace these with your own sensor reads
    return {
        "temperature": 23.5,
        "humidity": 60,
        "pressure": 1013.25,
    }


# Every call publishes the result; no manual publish() needed
while True:
    read_sensors()
    time.sleep(10)

Your function stays clean. The SDK handles serialization, batching, and delivery.

Tether with Offline Backup#

Network goes down? Add write_offline=True and your data is stored locally in SQLite until connectivity returns:

python

@client.tether(tags=["critical"], write_offline=True, db_ttl=3600)
def critical_reading():
    return {"voltage": 3.3, "current": 0.5}

No data loss. No retry logic to write. The SDK handles it.

Receive Inbound Messages#

Handle messages sent back from Contact flows or other entities. The Python SDK has one receive mechanism: a callback passed to the constructor. It is called with every inbound message, and you branch on the message yourself.

python

def handle(message):
    tags = message.get("tags") or []
    if "diagnostic" in tags:
        client.publish({"self_test": "pass"}, tags=["diagnostic-result"])
    elif "ai-response" in tags:
        print("AI:", message.get("data", {}).get("response"))
    else:
        print("Unhandled:", message.get("msg_type"), tags)

client = Client(api_key="your_api_key", callback=handle)
client.start()

The callback must be set on the constructor — polling only runs when one is present, and there is no @client.on() decorator or tag-based router in this SDK. (Go and MicroPython have one; see the SDK overview.) A callback that raises is logged and skipped, so one bad message cannot stop the client.

Tune polling with check_msg_rate (seconds between polls) and check_msg_limit (messages per poll).

Track Device State#

Every entity has a persistent state table, but the Python SDK does not wrap it — there is no @client.on_state() decorator and no read or write method. Call the REST endpoints directly with the entity's API key:

Method Path Purpose
GET /api/entities/status-table Read current state
PATCH /api/entities/status-table Merge into existing state
PUT /api/entities/status-table Replace state entirely

See State Table for full examples, and Examples for a working httpx snippet.

What You Get for Free#

When you call client.start(), the SDK automatically:

The SDK does not send heartbeats — it has no heartbeat support at all — and it does not retry an individual failed send with backoff. Delivery on a bad network depends on offline_storage; see below.

Knowing When Messages Don't Arrive#

publish() returns before the message is sent, so it cannot report a delivery failure. Instead the SDK logs to a standard library logger named tendrl, at most once a minute:

python


logging.basicConfig(level=logging.WARNING)
logging.getLogger("tendrl").setLevel(logging.WARNING)

With no logging configured at all, Python's last-resort handler still puts these warnings on stderr. There are two of them:

If you see neither, messages are arriving. Treat the first one as a paging alert: it is the only signal that data is being lost.

Complete Example#

python


from tendrl import Client

logging.basicConfig(level=logging.WARNING)  # surfaces undelivered-message warnings


def on_message(message):
    if "ai-response" in (message.get("tags") or []):
        print("AI:", message.get("data", {}).get("response"))


client = Client(
    api_key="your_api_key",
    offline_storage=True,
    callback=on_message,
)


@client.tether(tags=["sensor"], write_offline=True)
def collect():
    return {"temperature": 23.5, "humidity": 60}


client.start()

try:
    while True:
        collect()
        time.sleep(10)
except KeyboardInterrupt:
    client.stop()

Operating Modes#

Direct API Mode (Default)#

python

client = Client(mode="api", api_key="your_key")

Communicates directly with Contact over HTTP/2. Best for development and moderate throughput.

Nano Agent Mode#

python

client = Client(mode="agent")

Routes messages through a local Nano Agent Unix socket. Best for production and high throughput (50+ msg/sec). Requires the Nano Agent running locally.

Headless Mode#

python

client = Client(api_key="your_key", headless=True)

No background threads. Every publish() sends immediately and returns the response. Use for simple scripts or one-off sends.

What's Next#