Integrating a New Machine

Each machine in PUDA runs an edge service. The edge service wraps the machine's hardware API, connects to NATS, and exposes machine commands to the rest of the PUDA system.

info

PUDA edge services currently support Python only, because it is the most commonly used language in the lab automation ecosystem. If you prefer to use another language, contact us with your needs and we can create a PUDA SDK for it.

Use the edge-python-template template to create a new repository. Name it puda-<machine_id>-edge, replacing <machine_id> with your machine ID. The template depends on the puda Python SDK from PyPI. Require 0.0.17 or later for this CLI release:

pip install "puda>=0.0.17"

git clone https://github.com/<your-org>/puda-<machine_id>-edge.git
cd puda-<machine_id>-edge

Then follow the instructions in the README.md file.

Multiple machines on one PC

If several instruments share one operator PC, you do not need a separate repository for each edge service. Combine them into one repo and give each instrument its own subdirectory with its own MACHINE_ID, driver, and main.py.

PUDAP/ViPSA is a working example: a single repo with a uv workspace that runs stage, Zaber, light, and Keithley edge services from one machine.

Structure the repo like this:

puda-<name>/
├── pyproject.toml          # uv workspace — lists each edge as a member
├── uv.lock
├── stage/
│   ├── .env                # MACHINE_ID=stage
│   ├── driver.py
│   └── main.py
├── zaber/
│   ├── .env                # MACHINE_ID=zaber
│   ├── driver.py
│   └── main.py
└── light/
    ├── .env                # MACHINE_ID=light
    ├── driver.py
    └── main.py

Declare every edge package in the root pyproject.toml:

pyproject.toml

[tool.uv.workspace]
members = [
    "stage",
    "zaber",
    "light",
]

Each subdirectory is a standalone edge service from the edge-python-template template. Give each one a unique MACHINE_ID in its .env file, then start them individually or with a small script that launches every edge process on the PC.

Driver class

The main integration point is the driver.py class. This class should be a thin wrapper around the hardware code for your machine to make it more AI ready. If you already have example code for each command, an AI coding agent can usually turn that into a quick driver.py for you.

Only methods marked @command are advertised and callable. Undecorated public methods stay local to the driver. Drivers with no @command methods fail until the edge uses Python SDK 0.0.17 or later. Do not use types JSON cannot represent in @command parameters; see JSON primitives.

puda machine commands <machine_id>

caution

Keep every machine action you want an AI agent to discover, call, and orchestrate on this driver wrapper class, and mark it with @command. Methods that live only in lower-level hardware clients or helper classes are not available as machine commands.

This means the driver method names, docstrings, parameters, and return annotations become part of the command interface that users and agents see. Optional @safety metadata is published in the command catalog so agents can prompt the operator before execution; it does not change dispatch. See @safety decorator for an example.

Put a one-sentence summary on the Driver class docstring. PUDA advertises the first paragraph on puda machine list and puda machine ping so agents can tell what the machine does without fetching the command catalog.

driver.py

class Driver:
    """Software-only PUDA test machine. Commands are simulated in memory; no hardware is required."""

That first paragraph is what agents see:

puda machine list

{
  "machines": [
    {
      "machine_id": "test-1",
      "description": "Software-only PUDA test machine. Commands are simulated in memory; no hardware is required."
    }
  ],
  "count": 1
}

Best practices

When writing the driver class:

• Add a class docstring whose first paragraph is a one-sentence summary of what the machine does.
• Mark remotely callable methods with @command. Undecorated methods are not advertised.
• Attach @safety to commands that can cause harm so agents check preconditions and optionally confirm with the operator first.
• Add docstrings for every @command method to provide context.
• Document each parameter, error thrown and return object in the docstring.
• Keep commands atomic, with each method performing one clear machine action.
• Annotate @command parameters with JSON primitives. Protocol validate type-checks those annotations on commands the protocol uses. Unannotated parameters skip the type check. Types JSON cannot represent still fail validate when that command is used.
• Avoid exposing low-level helper methods as @command methods. Use private methods for internal helpers.
• Raise an exception when a command fails. Returning False is still a successful PUDA response.
• Lifecycle methods are optional: shutdown (called before the edge stops), home (triggered by puda machine home), and reset (puda machine reset always clears the run first).

These are standard good programming practices. In PUDA, they also become useful context for AI agents because the command names, docstrings, types, and errors describe what the machine can do and how failures should be interpreted.

JSON primitives

puda protocol validate checks each used command's params against the live catalog. Parameter types must be JSON primitives because protocol files and command payloads are JSON, so the catalog can only describe types JSON can carry.

Validate only parses catalog entries named in the protocol. Unused signatures are ignored, so an unparseable type on a command you are not running cannot fail the protocol.

Unannotated parameters, *args, and **kwargs skip the type check (any). *args is not sent over NATS (dispatch is kwargs-only). Annotate parameters you want checked.

Catalog kind	JSON value	Example annotation
str	string	name: str
int	integer	count: int
float	number	x_mm: float
bool	boolean	enabled: bool
bytes	string	blob: bytes
null	null	None
dict	object	layout: dict[str, str]
list	array	values: list[int]
any	any JSON value	payload: Any

Nullable values use T | None. Unions of primitives such as int | float are allowed. Object keys must be str. Nested objects and arrays of primitives are allowed, for example dict[str, str] or list[int].

driver.py

from puda import command

class Driver:
    """Example machine whose commands use JSON primitives."""

    @command
    def load_deck(self, layout: dict[str, str]) -> dict[str, str | None]:
        """Load labware names onto deck slots."""
        ...

Lifecycle commands

These lifecycle methods are optional. Define them when the machine has matching hardware behavior.

• shutdown: if defined, the edge calls it before the process stops so ports, sockets, and other resources are released.
• home: if defined, triggered by puda machine home <machine_id>. Omit it if the machine has no home motion.
• reset: if defined, puda machine reset <machine_id> calls it after clearing the active run ID. The CLI command still succeeds and clears the run if reset is omitted.

driver.py

from puda import command

class Driver:
    """Software-only PUDA test machine. Commands are simulated in memory; no hardware is required."""

    @command
    def shutdown(self) -> bool:
        """
        Shutdown the machine. If defined, called before the edge stops running.

        Returns:
            bool: True if the shutdown was successful, False otherwise
        """
        return True

    @command
    def home(self) -> bool:
        """
        Homes the machine. If defined, triggered by PUDA CLI `puda machine home <machine_id>`.

        Returns:
            bool: True if the home was successful, False otherwise
        """
        return True

    @command
    def reset(self) -> bool:
        """
        Software reset the hardware. If defined, `puda machine reset <machine_id>`
        calls this after clearing the active run ID.

        Returns:
            bool: True if the reset was successful, False otherwise
        """
        return True

Command design

Prefer atomic commands that map to a single hardware action. Avoid commands that combine unrelated work, hide required inputs, or require complex objects that cannot be represented cleanly as JSON primitives in the protocol file.

When a hardware operation fails, include the command, target values, and hardware response in the error message.

@safety decorator

Attach @safety to a @command method to publish advisory context in the command catalog. It does not block edge dispatch. @safety without @command is ignored. Use JSON primitives for @safety parameters.

All five fields are keyword-only:

Field	Type	Required	Role
summary	str	yes	Headline of the risk
hazards	list[str]	no	Short tags such as collision or thermal
requires	str	no	Preconditions; agents insert earlier steps or ask first
forbidden_when	str	no	Situations where the command must not be used
confirm	bool	no	Defaults to false. When true, the AI agent asks the operator to confirm before running the command.

Use requires for checks the agent must run first, such as a vision pass before motion or reagent checks before mixing:

driver.py

from puda import command, safety

@command
@safety(
    summary="Collision risk from moving into an occupied workspace.",
    hazards=["collision"],
    requires="Before running this command, use the vision analyse skill to check the workspace for obstructions.",
    forbidden_when="Do not move if vision reports an obstruction in the path.",
    confirm=True,  # manual confirmation needed
)
def move_to(self, x_mm: float, y_mm: float, z_mm: float) -> bool:
    """
    Move the machine head to an absolute position.

    Args:
        x_mm: Target X position in millimeters.
        y_mm: Target Y position in millimeters.
        z_mm: Target Z position in millimeters.

    Returns:
        bool: True if the move was accepted, False otherwise.

    Raises:
        RuntimeError: If the hardware rejects the move command.
    """
    result = self._hardware.move_to(x_mm=x_mm, y_mm=y_mm, z_mm=z_mm)
    if not result.ok:
        raise RuntimeError(
            "Failed to move machine head "
            f"to x={x_mm}, y={y_mm}, z={z_mm}: {result.error}"
        )
    return True

The agent should run the vision analyse skill, confirm the path is clear, then call move_to. Because confirm=True, the agent then prompts the operator for confirmation before it proceeds with the move.

Configure the edge service

After the driver commands work, edit main.py so the edge service loads the machine-specific environment variables, passes them into the driver, and publishes useful telemetry.

Keep MACHINE_ID and NATS_SERVERS in .env, then add the variables your hardware needs:

.env

MACHINE_ID=my-machine
NATS_SERVERS=nats://<nats-server-host>:<port>

# Add variables as needed
MY_MACHINE_HOST=192.168.1.50
MY_MACHINE_PORT=4000

Add matching fields to the Config class in main.py. Use descriptive names and types so missing or invalid configuration fails at startup instead of during a command:

main.py

class Config(BaseSettings):
    machine_id: str
    nats_servers: str
    my_machine_host: str
    my_machine_port: int = 4000

    model_config = SettingsConfigDict(
        env_file=Path(__file__).resolve().parent / ".env",
        env_file_encoding="utf-8",
        case_sensitive=False,
    )

    @property
    def nats_server_list(self) -> list[str]:
        return [s.strip() for s in self.nats_servers.split(",") if s.strip()]

Then instantiate the driver from that config:

main.py

driver = Driver(
    host=config.my_machine_host,
    port=config.my_machine_port,
)

The telemetry handler should publish at least heartbeat, optional position, health, and state.

main.py

async def telemetry_handler():
    await edge_nats_client.publish_heartbeat()

    await edge_nats_client.publish_position(driver.get_position())

    await edge_nats_client.publish_health({
        "connected": driver.is_connected(),
        "temperature_c": driver.temperature_c(),
    })


def state_handler():
    return {
        "state": driver.state,
        "run_id": driver.active_run_id,
        "homed": driver.is_homed(),
        "busy": driver.is_busy(),
    }


runner = EdgeRunner(
    nats_client=edge_nats_client,
    machine_driver=driver,
    telemetry_handler=telemetry_handler,
    state_handler=state_handler,
)

Adapt the field names to your machine. For a robot arm, position might include joints and tool pose. For a pump, position might be plunger position, valve position, and current flow rate. For a camera, it might be stage position, focus, exposure, and acquisition state. The important rule is that telemetry should be JSON-serializable, stable over time, and useful to agents, dashboards, and loggers watching subjects such as puda.<machine_id>.tlm.pos, puda.<machine_id>.tlm.health, and puda.<machine_id>.tlm.status.

Once this is done, you can install the PUDA CLI and skills in your AI agent so it can command your PUDA-integrated machines.