Extend Lumi’s capabilities by creating extensions or using ones created by other people.
Extensions can subscribe to Lumi data streams, expose tools, and provide information to the cloud agent that affects Lumi’s decision-making and planning.
A simple extension might integrate another system, such as ordering food through DoorDash’s CLI or accessing an email account. An extension can also coordinate new hardware—see our example below.
How extensions work
An extension is a program that runs on Lumi’s computer and communicates with Lightberry software through shared memory.
The extension SDK provides a simple interface for Python and Rust. To use another language, bind to the Rust SDK or communicate through iceoryx2 directly.
Lifecycle of an extension
sequenceDiagram
participant UP as User process
participant LB as Lightberry software
Note over UP: User process starts
UP->>LB: Register
LB-->>UP: Protocol version and capabilities
UP->>LB: Process information, tools, and subscriptions
LB-->>UP: Registration complete
Note over UP,LB: Operating state
LB->>UP: Shutdown notification
UP-->>LB: Acknowledge
Note over UP: User process shuts down- User process starts. The extension process starts inside Lumi’s environment.
- Register. The extension registers and receives the Lightberry protocol version and available capabilities.
- Advertise. The extension sends its process information, tools, and data subscriptions.
- Operate. Lightberry software and the extension exchange data and tool requests.
- Shut down. The host sends a shutdown notification and waits for an acknowledgement.
Example
This coffee dispenser extension connects to a coffee reservoir mounted on the robots back with a line and pump that dispenses from the center of the robot’s hand.
It exposes a dispense tool and periodically reports the coffee reservoir’s fill level to the planning agent.
//! Coffee dispenser extension that talks over serial to dispenser on the robot
//! hand fed from a reservoir mounted on the back.
use std::time::Duration;
use tokio::io::{AsyncBufReadExt as _, AsyncWriteExt as _, BufReader};
use tokio::time::{MissedTickBehavior, interval, timeout};
use tokio_serial::{
ClearBuffer, SerialPort as _, SerialPortBuilderExt as _, SerialStream,
};
use lightberry_extension_client::{
ExtensionAdvertisementT, ExtensionClient, HostToExtensionEventUnionT,
ToolDescriptionT, tool_response,
};
const SERIAL_PORT: &str = "/dev/ttyACM0";
const BAUD_RATE: u32 = 115_200;
const STATUS_INTERVAL: Duration = Duration::from_secs(1);
const STATUS_TIMEOUT: Duration = Duration::from_secs(2);
const DISPENSE_DESCRIPTION: &str = r"
Dispense 8 ounces of coffee from the dispenser mounted on the robot's hand.
Only use this tool when the hand is above an empty mug.
";
/// Newline-terminated ASCII coffee dispenser on a serial port.
struct CoffeeDispenser {
port: BufReader<SerialStream>,
}
impl CoffeeDispenser {
/// Open and configure the dispenser serial port.
fn open() -> Result<Self, Box<dyn std::error::Error>> {
let mut port =
tokio_serial::new(SERIAL_PORT, BAUD_RATE).open_native_async()?;
let _ = port.write_data_terminal_ready(true);
let _ = port.write_request_to_send(true);
let _ = port.clear(ClearBuffer::All);
Ok(Self {
port: BufReader::new(port),
})
}
/// Dispense 8 ounces of coffee.
async fn dispense(&mut self) -> Result<String, Box<dyn std::error::Error>> {
self.write_command("DISPENSE").await?;
Ok("Dispensed 8 ounces of coffee.".to_owned())
}
/// Read how full the dispenser is, from 0 to 100 percent.
async fn status(&mut self) -> Result<u32, Box<dyn std::error::Error>> {
let _ = self.port.get_mut().clear(ClearBuffer::Input);
self.write_command("STATUS").await?;
let mut line = String::new();
timeout(STATUS_TIMEOUT, self.port.read_line(&mut line)).await??;
let response = line.trim();
let level = response.parse::<u32>()?;
if level > 100 {
return Err(format!(concat!(
"dispenser STATUS reply {response:?} ",
"is not a percentage from 0 to 100"
))
.into());
}
Ok(level)
}
/// Write a command to the dispenser controller.
async fn write_command(
&mut self,
command: &str,
) -> Result<(), std::io::Error> {
self.port
.get_mut()
.write_all(format!("{command}\n").as_bytes())
.await?;
self.port.get_mut().flush().await
}
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let mut client =
ExtensionClient::connect_from_environment(ExtensionAdvertisementT {
instance_name: Some("coffee_dispenser".to_owned()),
display_name: Some("Coffee dispenser".to_owned()),
version: Some("0.0.1".to_owned()),
tools: Some(vec![ToolDescriptionT {
tool_name: Some("dispense".to_owned()),
description: Some(DISPENSE_DESCRIPTION.to_owned()),
..Default::default()
}]),
..Default::default()
})
.await?;
let mut dispenser = CoffeeDispenser::open()?;
let mut last_level = None;
let mut status_interval = interval(STATUS_INTERVAL);
status_interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
loop {
tokio::select! {
event = client.read_host_event() => {
match event?.event {
HostToExtensionEventUnionT::ToolInvocationRequest(
request,
) => {
let invocation_id = request.invocation_id.clone();
let response = match request.tool_name.as_deref() {
Some("dispense") => dispenser
.dispense()
.await
.map_err(|error| error.to_string()),
tool_name => Err(format!(
"unknown tool {}",
tool_name.unwrap_or("<missing>")
)),
};
client
.send_tool_invocation_response(tool_response(
invocation_id,
response,
))
.await?;
}
HostToExtensionEventUnionT::ShutdownRequest(_) => break,
HostToExtensionEventUnionT::NONE
| HostToExtensionEventUnionT::Ready(_)
| HostToExtensionEventUnionT::Data(_)
| HostToExtensionEventUnionT::ProtocolError(_) => {}
}
}
_ = status_interval.tick() => {
if let Ok(level) = dispenser.status().await
&& last_level != Some(level)
{
client
.send_extension_information_event(format!(
"Coffee dispenser is at {level}% full"
))
.await?;
last_level = Some(level);
}
}
}
}
Ok(())
}"""Coffee dispenser extension that talks over serial to dispenser on the robot
hand fed from a reservoir mounted on the back.
"""
import asyncio
from typing import Final
import serial_asyncio
from lightberry_extensions_sdk import (
ExtensionClient,
schema as extension,
tool_response,
)
SERIAL_PORT: Final = "/dev/ttyACM0"
BAUD_RATE: Final = 115_200
STATUS_INTERVAL_SECONDS: Final = 1.0
STATUS_TIMEOUT_SECONDS: Final = 2.0
DISPENSE_DESCRIPTION: Final = (
"Dispense 8 ounces of coffee from the dispenser mounted on the "
"robot's hand. "
"Only use this tool when the hand is above an empty mug."
)
async def _open_serial_connection(
serial_port: str, baud_rate: int
) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]:
"""Open an async serial connection."""
loop = asyncio.get_running_loop()
reader = asyncio.StreamReader()
protocol = asyncio.StreamReaderProtocol(reader)
transport, _ = await serial_asyncio.create_serial_connection(
loop, lambda: protocol, serial_port, baudrate=baud_rate
)
return reader, asyncio.StreamWriter(transport, protocol, reader, loop)
class CoffeeDispenser:
"""Newline-terminated ASCII coffee dispenser on a serial port."""
def __init__(
self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
) -> None:
self._reader = reader
self._writer = writer
@classmethod
async def open(
cls, serial_port: str = SERIAL_PORT, baud_rate: int = BAUD_RATE
) -> "CoffeeDispenser":
reader, writer = await _open_serial_connection(serial_port, baud_rate)
return cls(reader, writer)
async def close(self) -> None:
self._writer.close()
await self._writer.wait_closed()
async def dispense(self) -> str:
await self._write_command("DISPENSE")
return "Dispensed 8 ounces of coffee."
async def status(self) -> int:
serial_port = self._writer.transport.get_extra_info("serial")
if serial_port is not None:
serial_port.reset_input_buffer()
await self._write_command("STATUS")
line = await asyncio.wait_for(
self._reader.readline(), timeout=STATUS_TIMEOUT_SECONDS
)
if not line:
raise OSError("dispenser serial port closed")
response = line.decode().strip()
level = int(response)
if not 0 <= level <= 100:
raise ValueError(
f"dispenser STATUS reply {response!r} is not a percentage "
"from 0 to 100"
)
return level
async def _write_command(self, command: str) -> None:
self._writer.write(f"{command}\n".encode())
await self._writer.drain()
async def _publish_status(
client: ExtensionClient, dispenser: CoffeeDispenser, last_level: int | None
) -> int | None:
try:
level = await dispenser.status()
except (OSError, TimeoutError, ValueError):
return last_level
if level == last_level:
return last_level
await client.send_extension_information_event(
f"Coffee dispenser is at {level}% full"
)
return level
async def run() -> None:
"""Run the coffee dispenser extension."""
dispenser = await CoffeeDispenser.open()
client = await ExtensionClient.connect_from_environment(
extension.ExtensionAdvertisementT(
instanceName="coffee_dispenser",
displayName="Coffee dispenser",
version="0.0.1",
tools=[
extension.ToolDescriptionT(
toolName="dispense", description=DISPENSE_DESCRIPTION
),
],
)
)
last_level: int | None = None
host_event_task = asyncio.create_task(client.read_event())
status_task = asyncio.create_task(asyncio.sleep(0))
try:
while True:
done, _ = await asyncio.wait(
{host_event_task, status_task},
return_when=asyncio.FIRST_COMPLETED,
)
if status_task in done:
last_level = await _publish_status(
client, dispenser, last_level
)
status_task = asyncio.create_task(
asyncio.sleep(STATUS_INTERVAL_SECONDS)
)
if host_event_task not in done:
continue
event = host_event_task.result()
match event.eventType:
case (
extension.HostToExtensionEventUnion.ToolInvocationRequest
) if event.event is not None:
request = event.event
try:
match request.toolName:
case "dispense":
result: (
str | Exception
) = await dispenser.dispense()
case _:
raise ValueError(
f"unknown tool {request.toolName}"
)
except (OSError, TimeoutError, ValueError) as error:
result = error
await client.send_tool_invocation_response(
tool_response(request.invocationId, result)
)
host_event_task = asyncio.create_task(client.read_event())
case extension.HostToExtensionEventUnion.ShutdownRequest:
break
case _:
host_event_task = asyncio.create_task(client.read_event())
finally:
host_event_task.cancel()
status_task.cancel()
await client.close()
await dispenser.close()
if __name__ == "__main__":
asyncio.run(run())