Building a Server in Python

The MeshSocket repo ships a Python client (socketCore.MeshSocket) and a relay (socket_server.MeshServer). This is the lowest-effort way to integrate. For other languages, see Building a Server in Any Language (Raw Protocol); for the concepts, Data Flow — sync & actions.

Getting the library

The client lives in the MeshSocket repo under Python/ (socketCore.py, socket_server.py). It is not yet published to PyPI, so vendor it or add it to your path:

import sys
sys.path.insert(0, "/path/to/MeshSocket/Python")
from socketCore import MeshSocket          # the client
from socket_server import MeshServer       # the relay (for self-hosting)

Dependencies: websockets (the only hard requirement for the client).

The client API

MeshSocket(
    url="ws://host:8765",     # relay URL
    name="My Server",         # display name / routing handle
    auth_token="",            # gateway token; "" for self-hosted
    channel="home",           # channel to join
    role="hub",               # role (sets capability defaults)
    can_broadcast=True,       # may send broadcast_request
    can_route=True,           # may send route_msg / route_msg_noreply
    can_monitor=False,        # receive roster / call get_nodes
)

A handler that returns a value answers a request (the library wraps it into a reply_to); returning None is fire-and-forget.

Minimal server: push telemetry + handle a command

import asyncio, random, sys
sys.path.insert(0, "/path/to/MeshSocket/Python")
from socketCore import MeshSocket

RELAY   = "ws://192.168.1.50:8765"   # your relay; wss + auth_token for Connect+
CHANNEL = "home"

socket = MeshSocket(url=RELAY, name="Demo Server", channel=CHANNEL,
                    role="hub", can_broadcast=True, can_route=True)

# --- device → server: actions arrive as `broadcast`, demux on msg_type ---
@socket.on("broadcast")
async def on_broadcast(payload):
    if not isinstance(payload, dict):
        return
    if payload.get("msg_type") == "command":
        print("command:", payload.get("command"), payload)
    elif payload.get("msg_type") == "set":
        print("set:", payload.get("target"), "→", payload.get("value"))
    # no return value → fire-and-forget

# --- server → device: broadcast metrics every second ---
async def telemetry_loop():
    while True:
        await socket.emit("broadcast_request", {
            "msg_type": "metrics",
            "cpu": random.randint(0, 100),
            "mem": random.randint(0, 100),
        })
        await asyncio.sleep(1)

async def main():
    asyncio.create_task(socket.start())
    await socket.wait_until_ready()
    print("connected to", RELAY, "on", CHANNEL)
    await telemetry_loop()

asyncio.run(main())

Pair this with the tiny layout (gauge filtered on msg_type: "metrics", button broadcasting msg_type: "command") and the gauge moves while taps print.

Answering a request (routed RPC)

When a control uses mode: "request" (e.g. a graph node fetching content via route_msg), the frame is routed to you as a request of the inner type. Register a handler that returns the response object:

@socket.on("get_doc")
async def on_get_doc(payload):
    path = (payload or {}).get("path", "")
    return { "title": path, "content": load_markdown(path) }   # becomes the reply

See Recipe: Graph + Request/Reply Content for the matching layout side.

Running a relay (self-hosting)

For LAN use, run the bundled relay — no gateway, no token (it accepts any client when MESH_AUTH_TOKEN is unset):

import asyncio, sys
sys.path.insert(0, "/path/to/MeshSocket/Python")
from socket_server import MeshServer

# Binds 0.0.0.0:8765; point both your server and the phone's QR/layout at
# ws://<this-machine-LAN-IP>:8765
asyncio.run(MeshServer(host="0.0.0.0", port=8765).start())

MeshServer also accepts auth_handler=lambda token, ip: ... if you want simple token gating without the full Connect+ gateway, and on_startup / on_authenticated hooks. On macOS you may get a one-time firewall prompt to accept incoming connections — allow it.

Discovering the device (roster)

If you need to wait for a phone to appear (e.g. before pushing), poll rather than rely on the membership push:

async def wait_for_peer(timeout=30):
    import time; deadline = time.time() + timeout
    while time.time() < deadline:
        resp = await socket.request("get_nodes", timeout=3.0)
        peers = [c for c in (resp or {}).get("clients", []) if c["name"] != socket.name]
        if peers:
            return peers
        await asyncio.sleep(1)
    return []

(The roster is pushed as server_client_list on changes, but that push can miss peers that join after you — get_nodes reflects the live roster at call time. See Message Reference.)

Production notes

• Connect+ gateway: use wss://… and pass a valid auth_token. Without a valid token you'll be parked at "awaiting gateway" (see Connection & Pairing).
• Reconnects: wrap start() so a dropped socket reconnects; the client supports offline buffering (max_offline_buffer) if you want sends to survive brief drops.
• Rate: a few broadcasts/sec/metric is plenty; the device coalesces UI updates and the gateway rate-limits.

Recipes: Recipe: Telemetry Dashboard, Recipe: Interactive Controls, Recipe: Graph + Request/Reply Content, Recipe: Server-Driven (Dynamic) UI.