BabyROS

Robot Communication

BabyROS is a lightweight, ROS-like communication framework built on Zenoh - Publisher/Subscriber for streaming data and Client/Server for request/response, with no message schemas or workspace setup to manage. Pick the pattern that matches what you're trying to build.

Publisher/Subscriber

Use Publisher/Subscriber when one node needs to stream data continuously (e.g. sensor readings) and one or more other nodes need to react to it in real time.

Goal: Create and run a Publisher and Subscriber node using Python.

Level: Beginner  ·  Time: ~2 minutes

Create a Publisher

Create a file named publisher_example.py:

"""
Zenoh Publisher Example
"""
import time
import babyros


if __name__ == "__main__":
    # The session is created automatically inside the Publisher
    imu_pub = babyros.node.Publisher(topic="imu")

    # Get list of topics in the session
    topics = babyros.get_topics_in_session()
    print("Active topics in current session:", topics)

    # Start publishing
    print("Starting sensor stream... (Press Ctrl+C to stop)")
    count = 0
    try:
        while True:
            data = {
                "acceleration": [0.1, 0.0, 9.8],
                "gyro": [0.0, 0.01, 0.0],
                "seq": [count]
            }

            imu_pub.publish(data=data)
            print(f"Sent seq: {count}")

            count += 1
            time.sleep(0.1)  # 10 Hz

    except KeyboardInterrupt:
        print("\n[Publisher] Interrupted by user.")
    finally:
        # CRITICAL: Close the Zenoh session gracefully
        imu_pub.delete()
        print("[Publisher] Cleanup complete.")

Create a Subscriber

Create a file named subscriber_example.py:

import time
import babyros

def log_imu(msg: dict):
    """
    Callback to print received IMU messages.
    """
    print(f"Received IMU data: Seq {msg['seq']} | Accel: {msg['acceleration']}")

if __name__ == "__main__":
    # Create a subscriber node for the "imu" topic
    sub = babyros.node.Subscriber(topic="imu", callback=log_imu)
    print("Subscriber created successfully!")

    try:
        while True:
            time.sleep(1)  # Keep script alive
    except KeyboardInterrupt:
        print("\n[Subscriber] Interrupted by user.")
    finally:
        sub.delete()  # Clean up resources
        print("Subscriber cleanup complete.")

Run the nodes

Open one terminal and run the publisher:

python publisher_example.py

You should see messages being published every 0.1 seconds.

Open a second terminal and run the subscriber:

python subscriber_example.py

The subscriber will print each IMU message as it receives it.

INFO

Press Ctrl+C in both terminals to stop the nodes.

Client/Server

Use Client/Server when one node needs to request something from another and wait for a reply - e.g. sensor calibration requests, control parameter queries, or cloud-edge communication.

Goal: Create and run a Client and Server node using Python.

Level: Beginner  ·  Time: ~15 minutes

Create a Server

Create a file named server_example.py:

import time
import babyros

def handle_request(request):
    """
    Example service callback.
    Args:
        request (dict | None): Data sent by the client. May be None if no payload.
    Returns:
        dict: The response sent back to the client.
    """
    if request is None:
        print("Callback: No request payload received.")
        return {"message": "No request received!"}

    print("Request received and processed!")
    return {
        "message": "Hello from server!",
        "received": request
    }

if __name__ == "__main__":
    server = babyros.node.Server("example/topic", handle_request)
    print("Server started successfully!")

    topics = babyros.get_topics_in_session()
    print("Active topics in current session:", topics)

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        print("Shutting down server...")
        server.delete()

Create a Client

Create a file named client_example.py:

import json
import babyros

if __name__ == "__main__":
    client = babyros.node.Client(topic="example/topic")

    topics = babyros.get_topics_in_session()
    print("Active topics in current session:", topics)

    request = {"param1": "value1", "param2": "value2"}
    print("Sending request:", json.dumps(request))

    response = client.request(data=request)

    if not response:
        print("No response received from server.")
    else:
        print("Response:", response[0]["received"])
        print("Request:", request)
        print("Equal?", request == response[0]["received"])

    client.delete()

Run the nodes

Open a terminal and start the server:

python server_example.py

The server will start listening for client requests.

Open a second terminal and run the client:

python client_example.py

You should see the request sent by the client and the server's response printed in both terminals.

INFO

Press Ctrl+C in both terminals to stop the nodes.

Free Tier

Every new account starts on a free tier with credits to call the SDK — no billing details required.

Create API key →

Where to Go Next?

Ready to go deeper? Browse the Telekinesis Skill Examples repository for a runnable example covering every Skill.

Skill Examples

One runnable example per Skill, across vision, 3D, hardware, and robotics.

Explore →

Explore the Docs

Skills

Vision, 3D, hardware, and robotics skills — the full API reference.

Explore →

Agents

Turn natural-language instructions into robot code with Tzara, the VS Code agent.

Explore →

Data Engine

Generate and manage synthetic datasets for training Physical AI models.

Explore →

BabyROS

Lightweight pub/sub and service primitives for building robot communication layers.

Explore →

Applications

Real-world use cases: pick-and-place, palletizing, quality inspection, and more.

Explore →

Support

Join Discord

Ask questions, share what you're building, and get help from the community.

Join →

Add a Skill?

Have a new Skill to be added to Telekinesis? Get in touch with us.

Get in touch →