Tendrl

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, offline storage, heartbeats, and reconnection, 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/python-sdk

or with plain pip:

bash

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

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. Run with Client(debug=True) during setup to print those send failures.

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():
    return {
        "temperature": read_temp(),
        "humidity": read_humidity(),
        "pressure": read_pressure()
    }

# 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.

Route Inbound Messages

Handle messages sent back from Contact flows or other entities. Route by tag or msg_type with @client.on():

python

client = Client(api_key="your_api_key")

@client.on(tag="diagnostic")
def run_diagnostics(message):
    report = self_test(message.get("data", {}))
    client.publish(report, tags=["diagnostic-result"])

@client.on(tag="ai-response")
def handle_ai_reply(message):
    print("AI:", message.get("data", {}).get("response"))

@client.on(tags=["alert", "anomaly"])
def handle_alert(message):
    print("Alert:", message.get("data"))

client.start()

Register specific routes before broad ones, because the first match wins. Pass callback= as a catch-all fallback for unmatched messages.

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. Receive changes with @client.on_state(), which is polled at the same interval as messages:

python

@client.on_state()
def on_remote_state(state):
    if state.get("status") == "needs_maintenance":
        run_diagnostics()

To read or write state on demand, use the REST endpoints (see State Table).

What You Get for Free

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

You don't configure any of this. It just works.

Complete Example

python

from tendrl import Client

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

@client.on(tag="ai-response")
def on_ai_response(message):
    print("AI:", message.get("data", {}).get("response"))

@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

← Back to Contact docs