FostromDocs
ON THIS PAGE

Python SDK
PyPI package

The Fostrom Device SDK for Python is built to work on Linux and macOS. We test it across all architectures and operating systems we support. The minimum Python version required is 3.10.

Installation

You can install the package using pip:

pip install fostrom            # or `uv add fostrom`

We recommend using uv to manage your Python and pip installation and environment. Follow the steps below to setup a new project with uv and fostrom.

Setting up with uv

The simplest way to get up and running with the Fostrom Device SDK in Python is to use uv. Just initialize the project and add the dependency.

mkdir device && cd device   # Create project folder

uv init                     # Initialize a new Python project

uv venv                     # Create a Python virtual environment

source .venv/bin/activate   # Activate the virtual environment

uv add fostrom              # Fetch the `fostrom` python package

Then write your code in main.py. Check out the Usage section. You can connect some sensors to send datapoints to Fostrom. After that, just run your Python program using uv:

uv run main.py

Usage

Since Python does not support running scripts at install time, the Fostrom Device Agent is downloaded the first time you run your program while importing the library. Note that you need to have either curl or wget installed on your system to download it.

from fostrom import Fostrom, Mail



# Create SDK instance

fostrom = Fostrom({

    "fleet_id": "<your-fleet-id>",

    "device_id": "<your-device-id>",

    "device_secret": "<your-device-secret>",

    "env": "<dev|test|prod>",

})



# Setup mail handler for incoming messages

def handle_mail(mail: Mail):

    print(f"Received: {mail.name} ({mail.id})")



    # Fields:

    # * `mail.name`: packet schema name

    # * `mail.id`: a unique ID of this mail

    # * `mail.payload`: an optional payload



    # Acknowledge the message

    mail.ack() # or `mail.reject()`, or `mail.requeue()`



# Attach the mail handler

fostrom.on_mail = handle_mail



# Connect to Fostrom

fostrom.start()



# Send a Datapoint

fostrom.send_datapoint("<packet-schema-name>", {...payload})



# Send a Message

fostrom.send_msg("<packet-schema-name>", {...payload})

You can repeatedly read some sensor value in a loop and send datapoints and messages to Fostrom, or keep the program alive to process incoming mail.

Logging

The SDK logs via Python's logging module using the fostrom logger. By default, WARNING/ERROR/CRITICAL are visible; INFO is not.

To see INFO messages (e.g., "Connected"), enable logging:

import logging



logging.basicConfig(level=logging.INFO)

About the Device Agent

The Fostrom Device SDK downloads and runs the Fostrom Device Agent the first time your program is run with the import statement. The Device Agent starts when fostrom.start() is called and handles all communication with the Fostrom platform. The Device Agent runs continuously in the background, even after your Python program has terminated, to ensure that when your process manager restarts your Python program, it connects to the Device Agent right away. In case you want to stop the Device Agent when your program terminates, you can call fostrom.shutdown(True).

SDK API Reference

 

# Create a new Fostrom instance

fostrom = Fostrom(config: {

  fleet_id: str,                            # Fleet ID

  device_id: str,                           # Device ID

  device_secret: str,                       # Device Secret

  log: bool, default: True,                 # Enable/Disable Logging

  env: str, default: PYTHON_ENV || "dev",   # Runtime Env: dev | test | prod

})



# Start the Device Agent and connect to Fostrom

fostrom.start() -> None



# Disconnect from the Device Agent.

# If True is passed, the Device Agent is also shutdown.

fostrom.shutdown(stop_agent: bool = False) -> None



# Send a datapoint to Fostrom. A payload is required.

fostrom.send_datapoint(name: str, payload: dict) -> None



# Send a message to Fostrom. Payloads are optional.

fostrom.send_msg(name: str, payload: dict | None) -> None



# Get the mailbox status manually.

fostrom.mailbox_status() -> {

  mailbox_size: int,

  next_mail_id: str,

  next_mail_name: str

}



# Get the next mail from the mailbox.

fostrom.next_mail() -> Mail | None



# Note that both mailbox_status() and next_mail() are provided in the API,

# but do not need to be used. You should instead define the mail handler

# as shown above and process incoming mails inside it.



Mail{}                      # The Mail class:

  id: str                 # Mail ID (a UUIDv7 ID string)

  name: str               # Packet Schema Name

  payload: dict | None    # An optional payload

  mailbox_size: int       # Current Mailbox Size (including this mail)



  ack()                   # Acknowledge the mail

  reject()                # Reject the mail

  requeue()               # Requeue the mail