DimOS Modules

Modules are subsystems on a robot that operate autonomously and communicate with other subsystems using standardized messages. Some examples of modules are:

Webcam (outputs image)
Navigation (inputs a map and a target, outputs a path)
Detection (takes an image and a vision model like YOLO, outputs a stream of detections)

Below is an example of a structure for controlling a robot. Black blocks represent modules, and colored lines are connections and message types. It’s okay if this doesn’t make sense now. It will by the end of this document.

Prerequisite: Blueprint visualization (both SVG export and the Rerun Graph tab) requires Graphviz:
sudo apt install graphviz   # Ubuntu/Debian
brew install graphviz        # macOS

skip output=assets/go2_nav.svg

from dimos.core.introspection.svg import to_svg
from dimos.robot.unitree.go2.blueprints.smart.unitree_go2 import unitree_go2

to_svg(unitree_go2, "assets/go2_nav.svg")

Camera Module

Let’s learn how to build stuff like the above, starting with a simple camera module.

skip session=camera_module_demo output=assets/camera_module.svg

from dimos.hardware.sensors.camera.module import CameraModule
from dimos.core.introspection.svg import to_svg
to_svg(CameraModule.module_info(), "assets/camera_module.svg")

We can also print Module I/O quickly to the console via the .io() call. We will do this from now on.

session=camera_module_demo ansi=false

from dimos.hardware.sensors.camera.module import CameraModule

print(CameraModule.io())

┌┴─────────────┐
│ CameraModule │
└┬─────────────┘
 ├─ color_image: Image
 ├─ camera_info: CameraInfo
 │
 ├─ RPC build() -> None
 ├─ RPC get_skills() -> list
 ├─ RPC peek_stream(stream_name: str, timeout: float) -> Any
 ├─ RPC set_module_ref(name: str, module_ref: RPCClient) -> None
 ├─ RPC set_transport(stream_name: str, transport: Transport) -> bool
 ├─ RPC start() -> None
 ├─ RPC stop() -> None
 ├─ RPC take_a_picture() -> Image

We can see that the camera module outputs two streams:

color_image with sensor_msgs.Image type
camera_info with sensor_msgs.CameraInfo type

It offers two RPC calls: start() and stop() (lifecycle methods). It also exposes an agentic skill called take_a_picture (more on skills in the Blueprints guide). We can start this module and explore the output of its streams in real time (this will use your webcam).

skip session=camera_module_demo ansi=false

import time

camera = CameraModule()
camera.start()
# Now this module runs in our main loop in a thread. We can observe its outputs.

print(camera.color_image)

camera.color_image.subscribe(print)
time.sleep(0.5)
camera.stop()

Out color_image[Image] @ CameraModule
Image(shape=(480, 640, 3), format=RGB, dtype=uint8, dev=cpu, ts=2025-12-31 15:54:16)
Image(shape=(480, 640, 3), format=RGB, dtype=uint8, dev=cpu, ts=2025-12-31 15:54:16)
Image(shape=(480, 640, 3), format=RGB, dtype=uint8, dev=cpu, ts=2025-12-31 15:54:17)
Image(shape=(480, 640, 3), format=RGB, dtype=uint8, dev=cpu, ts=2025-12-31 15:54:17)
Image(shape=(480, 640, 3), format=RGB, dtype=uint8, dev=cpu, ts=2025-12-31 15:54:17)
Image(shape=(480, 640, 3), format=RGB, dtype=uint8, dev=cpu, ts=2025-12-31 15:54:17)
Image(shape=(480, 640, 3), format=RGB, dtype=uint8, dev=cpu, ts=2025-12-31 15:54:17)
Image(shape=(480, 640, 3), format=RGB, dtype=uint8, dev=cpu, ts=2025-12-31 15:54:17)
Image(shape=(480, 640, 3), format=RGB, dtype=uint8, dev=cpu, ts=2025-12-31 15:54:17)
Image(shape=(480, 640, 3), format=RGB, dtype=uint8, dev=cpu, ts=2025-12-31 15:54:17)

Connecting modules

Let’s load a standard 2D detector module and hook it up to a camera.

skip ansi=false session=detection_module

from dimos.perception.detection.module2D import Detection2DModule, Config
print(Detection2DModule.io())

 ├─ color_image: Image
┌┴──────────────────┐
│ Detection2DModule │
└┬──────────────────┘
 ├─ detections: Detection2DArray
 ├─ annotations: ImageAnnotations
 ├─ detected_image_0: Image
 ├─ detected_image_1: Image
 ├─ detected_image_2: Image
 │
 ├─ RPC build() -> None
 ├─ RPC get_skills() -> list
 ├─ RPC set_module_ref(name: str, module_ref: RPCClient) -> None
 ├─ RPC set_transport(stream_name: str, transport: Transport) -> bool
 ├─ RPC start() -> None
 ├─ RPC stop() -> None

Looks like the detector just needs an image input and outputs some sort of detection and annotation messages. Let’s connect it to a camera.

skip ansi=false

import time
from dimos.perception.detection.module2D import Detection2DModule, Config
from dimos.hardware.sensors.camera.module import CameraModule

camera = CameraModule()
detector = Detection2DModule()

detector.image.connect(camera.color_image)

camera.start()
detector.start()

detector.detections.subscribe(print)
time.sleep(3)
detector.stop()
camera.stop()

Distributed Execution

As we build module structures, we’ll quickly want to utilize all cores on the machine (which Python doesn’t allow as a single process) and potentially distribute modules across machines or even the internet. For this, we use dimos.core and DimOS transport protocols. Defining message exchange protocols and message types also gives us the ability to write models in faster languages.

Dedicated workers

By default the coordinator assigns modules to worker processes by least-load, so multiple modules share a worker. Heavy modules (robot connections, voxel mappers) should run alone so they don’t contend with anything else for CPU or the GIL. Set dedicated_worker = True on the class and the coordinator will give that module a worker process to itself.

from dimos.core.module import Module


class HeavyModule(Module):
    dedicated_worker = True

If declaring dedicated modules would push the pool past half-dedicated, the coordinator auto-grows it so non-dedicated workers always at least match the dedicated count.

Sync input handlers

If you don’t need an asyncio loop, subscribe to your In[T] streams from start() and register the unsubscribe with register_disposable so cleanup happens automatically at stop().

from reactivex.disposable import Disposable

from dimos.core.core import rpc
from dimos.core.module import Module
from dimos.core.stream import In
from dimos.msgs.std_msgs.Int32 import Int32


class Counter(Module):
    value: In[Int32]

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self._total = 0

    @rpc
    def start(self) -> None:
        super().start()
        self.register_disposable(Disposable(self.value.subscribe(self._on_value)))

    def _on_value(self, msg: Int32) -> None:
        self._total += msg.data

In.subscribe(cb) returns an unsubscribe function, not a DisposableBase. Wrap it in Disposable(...) so register_disposable can dispose it on stop(). Without this, your handler keeps running after stop() and tests will fail thread-leak checks. The callback runs on whatever thread emits the message, so guard mutable state with a lock if multiple inputs share it.

External Python Modules

An external Python module keeps its declaration in the host DimOS project and runs its implementation in a separately prepared local Python environment. The declaration is still an ordinary Blueprint participant: its typed streams, RPC methods, skills, configuration, and Spec module references are declared the same way as for any other Module. The only external-runtime setting is the implementation import reference:

from dimos.core.external_python_module import ExternalPythonModule


class MyFeature(ExternalPythonModule):
    implementation = "my_feature_runtime.runtime:MyFeatureRuntime"

Configuration belongs to the declaration contract and is inherited by the runtime implementation. Declare a module-specific Config(ModuleConfig), annotate the declaration’s config with it, and read it from the runtime:

from dimos.core.external_python_module import ExternalPythonModule
from dimos.core.module import ModuleConfig

class Config(ModuleConfig):
    initial_multiplier: int = 2

class MyFeature(ExternalPythonModule):
    config: Config

Blueprint callers override declaration configuration in the usual way: MyFeature.blueprint(initial_multiplier=3). The runtime reads the resolved value from self.config.initial_multiplier and may copy it into private mutable state during initialization. Restarting the module creates a fresh runtime, so that private state starts from the configured value again. Use the fixed sibling-project layout. The declaration source can have any filename; this example uses run.py:

external_python_module/
├── run.py
└── python/
    ├── pyproject.toml       # required
    ├── pixi.toml            # optional outer environment for uv
    └── my_feature_runtime/
        └── runtime.py

The source-checkout example keeps both runtime manifests in the sibling project. The example does not check in either a uv.lock or pixi.lock; both may be generated locally when preparing the runtime project and are ignored by git. uv and Pixi have separate lockfiles. The runtime class subclasses the declaration. Compose the declaration with regular modules using autoconnect; do not add a deployment plan, target, or special external-module API:

from examples.external_python_module.contract import ExampleExternal
from examples.external_python_module.run import ExampleConsumer
from dimos.core.coordination.blueprints import autoconnect

stack = autoconnect(ExampleExternal.blueprint(), ExampleConsumer.blueprint())

python/pyproject.toml is mandatory and must declare every Python dependency needed by the runtime, including its compatible DimOS package. In a source checkout, import the contract directly (for example, from examples.external_python_module.contract import ExampleExternal); the example is not part of the DimOS distribution. DimOS prepares and runs this project with uv. If python/pixi.toml is present, Pixi supplies the outer tool environment and runs uv; it does not replace the uv Python project or its dependency declarations. If the sibling python/pixi.toml is present, local preparation requires Pixi to be installed; Pixi is not supplied by DimOS. Preparation happens before the module starts. A missing python/ directory or pyproject.toml, a failed uv/Pixi preparation, an unavailable dependency, or an invalid implementation import reference aborts deployment with diagnostics from the failed step. A runtime that does not fulfill the declaration contract is rejected and cleaned up before the rest of the Blueprint starts. There is no remote deployment, package transfer, SSH, or deployment-specific CLI in this API; use the normal local Blueprint and coordinator lifecycle.

Triggering side effects via Specs

A common pattern is “subscribe to a stream, react by calling another module”. Declare the other module’s protocol as a Spec field (single-underscore, private). The coordinator binds the proxy at deploy time, so handlers can call it directly with no extra wiring:

from typing import Protocol

from reactivex.disposable import Disposable

from dimos.core.core import rpc
from dimos.core.module import Module
from dimos.core.stream import In
from dimos.msgs.std_msgs.Int32 import Int32
from dimos.spec.utils import Spec


class NotifierSpec(Spec, Protocol):
    def notify(self, text: str) -> None: ...


class Watchdog(Module):
    value: In[Int32]

    _notifier: NotifierSpec

    @rpc
    def start(self) -> None:
        super().start()
        self.register_disposable(Disposable(self.value.subscribe(self._on_value)))

    def _on_value(self, msg: Int32) -> None:
        if msg.data > 100:
            self._notifier.notify(f"value={msg.data}")

The Spec must match the target module’s @rpc signatures (sync/async are interchangeable — see Async modules). To deploy Watchdog, add Watchdog.blueprint() to an existing blueprint’s autoconnect(...) chain. The coordinator matches Out[T] to In[T] by name across the union of modules, and resolves _notifier: NotifierSpec to whichever module in the blueprint implements notify. No manual wiring required.

Testing modules

Mock spec dependencies (anything typed : SomeSpec) after construction, since the framework normally wires them at deploy time:

skip

@pytest.fixture()
def module(mocker):
    m = MyModule(step=10)
    m._speak_skill = mocker.MagicMock()
    yield m
    m.stop()  # required: cleans up the per-instance asyncio loop and thread

The m.stop() in teardown matters. The test session-wide thread-leak detector will fail the test otherwise, even if your test body never started any threads.

Restarting a module

While iterating on a module it’s often convenient to edit its source file and pick up the changes without tearing down the whole coordinator. The restart_module call stops a single deployed module, reloads its source via importlib.reload, then redeploys it onto a fresh worker process while keeping its stream transports and reconnecting any other modules that held a reference to it.

skip

from dimos.core.coordination.module_coordinator import ModuleCoordinator
from dimos.core.global_config import GlobalConfig
from dimos.hardware.sensors.camera.module import CameraModule

coordinator = ModuleCoordinator(g=GlobalConfig(n_workers=0, viewer="none"))
coordinator.start()
coordinator.load_module(CameraModule)

# ... edit CameraModule source on disk ...

coordinator.restart_module(CameraModule)

Async modules (lock-free state)

Modules contain a per-instance asyncio loop on a daemon thread (self._loop). It is possible to write modules using only async def methods so that everything runs on the same thread and you don’t need to use locks. The module’s auto-bound input handlers, async @rpc methods, and process_observable callbacks all run on self._loop, and each handler subscription is serialized through a dedicated dispatcher task.

Auto-bound input handlers

For every declared x: In[T], if the module defines async def handle_x(self, msg: T), the handler is automatically subscribed at start() and dispatched onto self._loop. Subscriptions are cleaned up at stop().

from dimos.core.module import Module
from dimos.core.stream import In, Out
from dimos.msgs.geometry_msgs.PointStamped import PointStamped
from dimos.msgs.geometry_msgs.Twist import Twist

class MovementManager(Module):
    clicked_point: In[PointStamped]
    nav_cmd_vel: In[Twist]
    tele_cmd_vel: In[Twist]

    cmd_vel: Out[Twist]
    goal: Out[PointStamped]

    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # No lock needed. `_teleop_active` is only mutated on `self._loop`.
        self._teleop_active = False

    async def handle_clicked_point(self, msg: PointStamped) -> None:
        self.goal.publish(msg)

    async def handle_nav_cmd_vel(self, msg: Twist) -> None:
        if not self._teleop_active:
            self.cmd_vel.publish(msg)

    async def handle_tele_cmd_vel(self, msg: Twist) -> None:
        self._teleop_active = True
        self.cmd_vel.publish(msg)

Each handler runs in a per-handler dispatcher task on self._loop. Handlers are serialized: only one invocation of handle_x runs at a time. If messages arrive faster than the handler can process them, intermediate messages are dropped — only the most recent unprocessed message is kept (LATEST policy). The handler is guaranteed to eventually run with the most recently published value.

Async `@rpc` methods

@rpc works on both sync and async def methods. When applied to an async method, the call site dispatches automatically:

From another thread (the RPC dispatcher, sync test code, a sync @rpc on the same module), the call blocks until the coroutine completes on self._loop.
From inside the loop (another async @rpc, a handle_*, or a process_observable callback), it returns the coroutine so the caller can await it.

from dimos.core.core import rpc
from dimos.core.module import Module

class NameModule(Module):
    @rpc
    async def say_hello(self, name: str) -> str:
        return f"Hello {name}, from {self._my_name}"

    @rpc
    async def set_my_name(self, new_name: str) -> None:
        self._my_name = new_name

Async and sync @rpc methods are interchangeable for cross-module linking. Both are discovered via Module.rpcs and served through the same RPC machinery. A module ref or RPC client doesn’t care whether the underlying method is sync or async. When the consumer types a module ref using a Spec that declares async def, the proxy automatically exposes those methods as awaitables: await self._name_module.say_hello(name).

from typing import Protocol

from dimos.core.module import Module
from dimos.spec.utils import Spec

class NameSpec(Spec, Protocol):
    async def say_hello(self, name: str) -> str: ...
    async def set_my_name(self, new_name: str) -> None: ...

class StartModule(Module):
    _name_module: NameSpec

    async def code():
        await self._name_module.set_my_name("John")
        print(await self._name_module.say_hello("Bill"))

NameModule is async. But if you need to call it from a sync module, you just need to create a SyncNameSpec:

from typing import Protocol

from dimos.spec.utils import Spec

class SyncNameSpec(Spec, Protocol):
    def say_hello(self, name: str) -> str: ...
    def set_my_name(self, new_name: str) -> None: ...

This will match with NameModule. You can call it synchronously from your module, but it will run in the self._loop async loop in the NameModule module. The reverse is also true: you can call a sync module from async code.

`spawn`: schedule a long-running coroutine from sync code

When you need to start a long-running async task from start() (e.g., a timer loop), use self.spawn(coro) instead of asyncio.run_coroutine_threadsafe(coro, self._loop). The helper wires up a done-callback that surfaces unhandled exceptions to the module logger. bare run_coroutine_threadsafe silently stores the exception on the returned Future, where it disappears unless the user remembers to read .result().

import asyncio

from dimos.core.core import rpc
from dimos.core.module import Module

class TimerExample(Module):
    @rpc
    def start(self) -> None:
        super().start()
        self._timer_future = self.spawn(self._timer_loop())

    async def _timer_loop(self) -> None:
        while True:
            await asyncio.sleep(1.0)
            ...

    @rpc
    def stop(self) -> None:
        if self._timer_future is not None:
            self._timer_future.cancel()
        super().stop()

`process_observable`: async subscriptions to arbitrary observables

Sometimes you have rxpy observables which you need to run inside self._loop. You can do this with self.process_observable(observable, async_handler) .

skip

@rpc
def start(self) -> None:
    super().start()
    fast = self.foo.observable().pipe(ops.filter(lambda v: v > threshold))
    self.process_observable(fast, self._on_fast_foo)

async def _on_fast_foo(self, v: int) -> None:
    ...

`main()`: combined setup/teardown

When a module owns a resource that needs construction at startup and explicit cleanup at shutdown, define async def main(self) as an async generator with exactly one yield. Code before yield runs at start(), code after yield runs at stop().

from collections.abc import AsyncIterator
from typing import Any

from dimos.core.module import Module

def create(name: str) -> Any:
    del name
    class _Model:
        def stop(self) -> None:
            pass

    return _Model()

class PersonFollowSkillContainer(Module):
    async def main(self) -> AsyncIterator[None]:
        # setup
        self._vl_model = create("qwen")

        yield

        # teardown
        self._vl_model.stop()

Compared to splitting the same work across __init__ / start() / stop(), main() keeps the construction-and-destruction of each resource visually adjacent.

Blueprints

A blueprint is a predefined structure of interconnected modules. You can include blueprints or modules in new blueprints. A basic Unitree Go2 blueprint looks like what we saw before.

skip session=blueprints output=assets/go2_agentic.svg

from dimos.core.introspection.svg import to_svg
from dimos.robot.unitree_webrtc.unitree_go2_blueprints import agentic

to_svg(agentic, "assets/go2_agentic.svg")

To see more information on how to use Blueprints, see Blueprints.

Getting Started

Installation

Usage

Capabilities

Platforms

Development

Camera Module

Connecting modules

Distributed Execution

Dedicated workers

Sync input handlers

External Python Modules

Triggering side effects via Specs

Testing modules

Restarting a module

Async modules (lock-free state)

Auto-bound input handlers

Async `@rpc` methods

`spawn`: schedule a long-running coroutine from sync code

`process_observable`: async subscriptions to arbitrary observables

`main()`: combined setup/teardown

Blueprints

​Camera Module

​Connecting modules

​Distributed Execution

​Dedicated workers

​Sync input handlers

​External Python Modules

​Triggering side effects via Specs

​Testing modules

​Restarting a module

​Async modules (lock-free state)

​Auto-bound input handlers

​Async @rpc methods

​spawn: schedule a long-running coroutine from sync code

​process_observable: async subscriptions to arbitrary observables

​main(): combined setup/teardown

​Blueprints

Camera Module

Connecting modules

Distributed Execution

Dedicated workers

Sync input handlers

External Python Modules

Triggering side effects via Specs

Testing modules

Restarting a module

Async modules (lock-free state)

Auto-bound input handlers

Async `@rpc` methods

`spawn`: schedule a long-running coroutine from sync code

`process_observable`: async subscriptions to arbitrary observables

`main()`: combined setup/teardown

Blueprints