Send telemetry data to Plexus using HTTP or WebSocket.

Two ways to send data:

Set up your device with one command using an API key:

# With API key (fleet provisioning — get from Settings → Developer)
curl -sL https://app.plexus.company/setup | bash -s -- --key plx_your_api_key

Then control streaming, recording, and configuration from app.plexus.company/devices.

Send data directly via HTTP:

curl -X POST https://plexus-gateway.fly.dev/ingest \
  -H "x-api-key: YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "points": [{
      "metric": "temperature",
      "value": 72.5,
      "timestamp": 1699900000,
      "source_id": "sensor-001"
    }]
  }'

Plexus uses API keys for all authentication:

Option A: CLI setup (recommended for devices)

1. Run plexus start on your device
2. Sign up or sign in directly in the terminal
3. API key is saved to ~/.plexus/config.json

Option B: Manual creation

1. Sign up at app.plexus.company
2. Go to Settings → Developer
3. Create an API key (starts with plx_)

All requests require an API key in the header:

x-api-key: plx_xxxxx

POST /api/ingest

{
  "points": [
    {
      "metric": "temperature",
      "value": 72.5,
      "timestamp": 1699900000.123,
      "source_id": "sensor-001",
      "tags": { "location": "lab" },
      "session_id": "test-001"
    }
  ]
}

Group related data for analysis and playback.

Create session:

POST /api/sessions
{
  "session_id": "test-001",
  "name": "Motor Test Run",
  "source_id": "sensor-001",
  "status": "active"
}

End session:

PATCH /api/sessions/{session_id}
{
  "status": "completed",
  "ended_at": "2024-01-15T10:30:00Z"
}

For real-time UI-controlled streaming, devices connect via WebSocket.

1. Device connects to PartyKit server
2. Device authenticates with API key
3. Device reports available sensors
4. Dashboard controls streaming via messages

Devices authenticate using an API key. The source_id in the request is the device's desired name; the server may return a different, auto-suffixed name in the authenticated frame if the desired name is already claimed by another device (see Device identity in the README).

// Device → Server
{
  "type": "device_auth",
  "api_key": "plx_xxxxx",
  "source_id": "drone-01",
  "install_id": "c9f2e0b46f4a4f6a8c3e1d5b0a2e7f91",
  "platform": "python-sdk",
  "agent_version": "0.3.1"
}

// Server → Device
{
  "type": "authenticated",
  "source_id": "drone-01",
  "server_time_ms": 1746100800000
}

// Server → Device (collision case)
{
  "type": "authenticated",
  "source_id": "drone-01_2",
  "server_time_ms": 1746100800000
}

The SDK adopts whatever source_id the server returns and uses it for all subsequent frames, heartbeats, and reconnects. It also persists the assigned name locally so reconnects go straight to the claimed slot.

server_time_ms is the gateway's current Unix time in milliseconds. The Python SDK uses it to compute a clock offset (server_time - device_time) that is applied to every SDK-generated timestamp for the lifetime of the connection. This corrects for devices that boot without NTP or have an unreliable RTC — a common condition on embedded Linux. See Clock correction for details and limitations.

install_id is a stable per-installation UUID, generated on the device's first run and saved to ~/.plexus/config.json. It lets the server distinguish a rebooting device from a new device trying to claim an existing name. Legacy SDKs that omit install_id continue to work as before (the server passes the declared source_id through unchanged).

// Dashboard → Device
{
  "type": "start_stream",
  "source_id": "my-device-001",
  "metrics": ["accel_x", "accel_y", "accel_z"],
  "interval_ms": 100
}

// Device → Dashboard (continuous)
{
  "type": "telemetry",
  "points": [
    { "metric": "accel_x", "value": 0.12, "timestamp": 1699900000123 },
    { "metric": "accel_y", "value": 0.05, "timestamp": 1699900000123 },
    { "metric": "accel_z", "value": 9.81, "timestamp": 1699900000123 }
  ]
}

// Dashboard → Device
{
  "type": "start_session",
  "source_id": "my-device-001",
  "session_id": "session_1699900000_abc123",
  "session_name": "Motor Test",
  "metrics": [],
  "interval_ms": 100
}

// Device → Dashboard
{
  "type": "session_started",
  "session_id": "session_1699900000_abc123",
  "session_name": "Motor Test"
}

// Device streams telemetry with session_id tag
{
  "type": "telemetry",
  "session_id": "session_1699900000_abc123",
  "points": [
    {
      "metric": "accel_x",
      "value": 0.12,
      "timestamp": 1699900000123,
      "tags": { "session_id": "session_1699900000_abc123" }
    }
  ]
}

// Dashboard → Device
{
  "type": "configure",
  "source_id": "my-device-001",
  "sensor": "MPU6050",
  "config": {
    "sample_rate": 50
  }
}

import requests
import time

requests.post(
    "https://plexus-gateway.fly.dev/ingest",
    headers={"x-api-key": "plx_xxxxx"},
    json={
        "points": [{
            "metric": "temperature",
            "value": 72.5,
            "timestamp": time.time(),
            "source_id": "sensor-001"
        }]
    }
)

await fetch("https://plexus-gateway.fly.dev/ingest", {
  method: "POST",
  headers: {
    "x-api-key": "plx_xxxxx",
    "Content-Type": "application/json",
  },
  body: JSON.stringify({
    points: [
      {
        metric: "temperature",
        value: 72.5,
        timestamp: Date.now() / 1000,
        source_id: "sensor-001",
      },
    ],
  }),
});

package main

import (
    "bytes"
    "encoding/json"
    "net/http"
    "time"
)

func main() {
    points := map[string]interface{}{
        "points": []map[string]interface{}{{
            "metric":    "temperature",
            "value":     72.5,
            "timestamp": float64(time.Now().Unix()),
            "source_id": "sensor-001",
        }},
    }

    body, _ := json.Marshal(points)
    req, _ := http.NewRequest("POST", "https://plexus-gateway.fly.dev/ingest", bytes.NewBuffer(body))
    req.Header.Set("x-api-key", "plx_xxxxx")
    req.Header.Set("Content-Type", "application/json")

    http.DefaultClient.Do(req)
}

#include <WiFi.h>
#include <HTTPClient.h>

// Call configTime(0, 0, "pool.ntp.org") in setup() before sending.
// time(nullptr) returns 0 until NTP sync completes — omit the timestamp
// field entirely if you cannot guarantee NTP sync at send time.
void sendToPlexus(const char* metric, float value) {
    HTTPClient http;
    http.begin("https://plexus-gateway.fly.dev/ingest");
    http.addHeader("Content-Type", "application/json");
    http.addHeader("x-api-key", "plx_xxxxx");

    String payload = "{\"points\":[{";
    payload += "\"metric\":\"" + String(metric) + "\",";
    payload += "\"value\":" + String(value) + ",";
    payload += "\"timestamp\":" + String(time(nullptr)) + ",";
    payload += "\"source_id\":\"esp32-001\"";
    payload += "}]}";

    http.POST(payload);
    http.end();
}

#!/bin/bash
API_KEY="plx_xxxxx"
SOURCE_ID="sensor-001"

curl -X POST https://plexus-gateway.fly.dev/ingest \
  -H "x-api-key: $API_KEY" \
  -H "Content-Type: application/json" \
  -d "{
    \"points\": [{
      \"metric\": \"temperature\",
      \"value\": 72.5,
      \"timestamp\": $(date +%s),
      \"source_id\": \"$SOURCE_ID\"
    }]
  }"

For MAVLink, CAN, MQTT, Modbus, OPC-UA, BLE, Serial, or any other protocol, use the upstream Python library directly and pass values to px.send(). Plexus stays out of your decode path:

# CAN example — using python-can directly
import can
from plexus import Plexus

px = Plexus(api_key="plx_xxx", source_id="vehicle-001")
bus = can.interface.Bus(channel="can0", bustype="socketcan")

for msg in bus:
    px.send(f"can.0x{msg.arbitration_id:x}", int.from_bytes(msg.data, "big"))

# MAVLink example — using pymavlink directly
from pymavlink import mavutil
from plexus import Plexus

px = Plexus(api_key="plx_xxx", source_id="drone-001")
conn = mavutil.mavlink_connection("udpin:0.0.0.0:14550")

while True:
    msg = conn.recv_match(blocking=True)
    if msg.get_type() == "ATTITUDE":
        px.send("attitude.roll", msg.roll)
        px.send("attitude.pitch", msg.pitch)

See docs.plexus.dev/recipes for more.

For Raspberry Pi and other Linux devices, the Python SDK includes sensor drivers:

pip install plexus-python[sensors]
plexus start

from plexus.sensors import BaseSensor, SensorReading

class MySensor(BaseSensor):
    name = "MySensor"
    metrics = ["voltage", "current"]

    def read(self):
        return [
            SensorReading("voltage", read_adc(0) * 3.3),
            SensorReading("current", read_adc(1) * 0.1),
        ]

Embedded devices commonly boot with a wrong system clock — no hardware RTC, NTP unreachable on first boot, or a fresh OS image whose filesystem timestamp is months in the past. Without correction, all telemetry lands at the wrong place on the timeline.

The Python SDK corrects for this automatically over WebSocket. On every connection the gateway returns server_time_ms in the authenticated frame. The SDK computes offset = server_time - device_time and adds it to every timestamp it generates. Data lands at the right time on the dashboard regardless of what the device clock says.

When the correction applies:

The offset is applied when timestamp is omitted (the SDK generates the time). If you pass an explicit timestamp, it is used as-is — the SDK cannot tell whether your value is a wall-clock time or a hardware-relative counter, so it leaves it alone.

px.send("temperature", 72.5)                    # SDK picks time → correction applied
px.send("temperature", 72.5, timestamp=t)        # your timestamp → used as-is, no correction

When to pass an explicit timestamp:

• You have a reliable wall-clock source (GPS, trusted hardware RTC, host NTP)
• You are replaying or backfilling historical data
• Your sensor provides its own wall-clock timestamp

When to omit timestamp:

• The device may have booted without NTP (Raspberry Pi, Jetson, field robots without network on first boot)
• You have no reliable external time source

Known limitations:

• The clock offset refreshes only on WebSocket reconnect. A device with a drifting RTC that stays connected for many days will accumulate uncorrected drift between reconnects proportional to the drift rate.
• HTTP transport (transport="http") does not receive clock sync — timestamps default to the device clock uncorrected.
• send_batch() takes one shared timestamp for the whole batch, not per-point. For per-point timestamps, call send() in a loop.

• Batch points - Send up to 100 points per request for HTTP
• Omit timestamp when unsure - The Python SDK applies server-synced clock correction when timestamp is omitted over WebSocket; only pass an explicit timestamp when you have a reliable wall-clock source
• Consistent source_id - Use the same ID for each physical device/source
• Use tags - Label data for filtering (e.g., {"location": "lab"})
• Use sessions - Group related data for easier analysis
• Prefer WebSocket - For real-time UI-controlled devices, use plexus start