madsci-nodes

name	madsci-nodes
description	Working with MADSci node modules for laboratory instrument integration. Use when creating, modifying, debugging, or testing node modules that interface with scientific instruments.

MADSci Node Modules

Node modules are the instrument integration layer of MADSci. Each node wraps a laboratory instrument (robot arm, plate reader, liquid handler, etc.) behind a standard API. The architecture is: AbstractNode (protocol-agnostic base) -> RestNode (FastAPI HTTP server). Both use MadsciClientMixin for lazy access to Event, Resource, and Data clients.

Key Files

File	Purpose
src/madsci_node_module/madsci/node_module/abstract_node_module.py	Base class with lifecycle, action dispatch, type analysis
src/madsci_node_module/madsci/node_module/rest_node_module.py	FastAPI server, file upload/download, rate limiting
src/madsci_node_module/madsci/node_module/helpers.py	@action decorator
src/madsci_node_module/madsci/node_module/type_analyzer.py	Recursive type hint analysis (TypeInfo)
src/madsci_common/madsci/common/types/node_types.py	NodeConfig, RestNodeConfig, NodeStatus, NodeInfo, NodeCapabilities, NodeIntrinsicLocationDefinition
src/madsci_client/madsci/client/node/rest_node_client.py	HTTP client for node communication
src/madsci_common/madsci/common/bundled_templates/node/basic/	madsci new node template

Creating a Node

1. Scaffold with template

madsci new node
# or: madsci new module  (full module with interface, types, tests, Dockerfile)

2. Minimal RestNode

from madsci.node_module.rest_node_module import RestNode
from madsci.node_module.helpers import action

class MyInstrumentNode(RestNode):
    """Node for controlling MyInstrument."""

    def startup_handler(self):
        """Initialize instrument connections."""
        self.instrument = MyInstrumentDriver()

    def shutdown_handler(self):
        """Clean up instrument connections."""
        self.instrument.disconnect()

    @action(name="measure", description="Take a measurement")
    def measure(self, sample_id: str, temperature: float = 25.0) -> dict:
        result = self.instrument.measure(sample_id, temp=temperature)
        return {"sample_id": sample_id, "value": result}

if __name__ == "__main__":
    node = MyInstrumentNode()
    node.start_node()

3. Configuration

Nodes use NodeConfig (AbstractNode) or RestNodeConfig (RestNode), both inheriting from MadsciBaseSettings:

# RestNodeConfig key fields:
node_name: Optional[str]       # Human name (defaults to class name)
node_id: Optional[str]         # ULID (auto-generated)
node_type: Optional[NodeType]  # DEVICE, COMPUTE, etc.
node_url: AnyUrl               # Default: http://127.0.0.1:2000
status_update_interval: float  # Default: 2.0s
state_update_interval: float   # Default: 2.0s
enable_registry_resolution: bool  # Default: True
enable_rate_limiting: bool     # Default: True

Config is discovered via walk-up file search (see CLAUDE.md). Override with environment variables using NODE_ prefix or pass directly:

node = MyNode(node_config=RestNodeConfig(node_url="http://0.0.0.0:3000"))

Deprecated: NodeDefinition

NodeDefinition is deprecated as of v0.7.0. Node identity now comes from NodeConfig/RestNodeConfig plus the ID Registry.

• Do not use NodeDefinition or *.node.yaml definition files in new code
• Use NodeConfig or RestNodeConfig (both are MadsciBaseSettings subclasses)
• Run madsci migrate to convert legacy *.node.yaml files

The @action Decorator

Register methods as node actions. Discovered automatically at init.

from madsci.node_module.helpers import action

# With explicit parameters
@action(name="dispense", description="Dispense liquid", blocking=True)
def dispense(self, volume_ul: float, well: str) -> dict:
    ...

# Minimal (name from function, description from docstring, blocking=True)
@action
def home(self):
    """Home all axes to reference position."""
    self.instrument.home()

Parameters:

• name (str, optional): Action name (defaults to function name)
• description (str, optional): Description (defaults to docstring)
• blocking (bool, default True): Whether the action blocks concurrent actions via _action_lock

Behind the scenes: Sets __is_madsci_action__ = True on the function. AbstractNode.__init__() scans __class__.__dict__ for this flag and auto-registers via _add_action().

Action Parameters and Return Types

Regular parameters

Standard Python types become JSON-serializable action arguments:

@action
def process(self, sample_id: str, count: int = 1, options: Optional[dict] = None) -> dict:
    ...

File parameters (use Path, not UploadFile)

from pathlib import Path

@action
def analyze_image(self, image: Path) -> dict:
    """Single file upload."""
    data = image.read_bytes()
    return {"size": len(data)}

@action
def batch_process(self, files: list[Path]) -> dict:
    """Multiple file upload."""
    return {"count": len(files)}

Location parameters

from madsci.common.types.location_types import LocationArgument

@action
def transfer(self, source: LocationArgument, destination: LocationArgument) -> dict:
    ...

Annotated parameters (extra metadata)

from typing import Annotated

@action
def calibrate(self, offset: Annotated[float, "Offset in mm, range -10 to 10"]) -> dict:
    ...

Return types

from madsci.common.types.action_types import ActionFiles, ActionDatapoints
from madsci.common.types.data_types import DataPoint

# Plain dict/Pydantic model -> json_result
@action
def measure(self) -> dict:
    return {"value": 42}

# Path -> single file result
@action
def capture(self) -> Path:
    return Path("/tmp/image.png")

# ActionFiles -> multiple named files
@action
def generate_report(self) -> ActionFiles:
    return ActionFiles(report=Path("report.pdf"), data=Path("data.csv"))

# Tuple -> (json_result, files, datapoints)
@action
def full_result(self) -> tuple[dict, ActionFiles, ActionDatapoints]:
    return (
        {"status": "ok"},
        ActionFiles(output=Path("out.csv")),
        ActionDatapoints(dp=DataPoint(label="measurement", value=42)),
    )

Node Lifecycle

__init__()
  ├── Load NodeConfig (walk-up discovery)
  ├── Synthesize NodeInfo
  ├── Resolve registry identity (if enabled)
  ├── Configure clients (EventClient, ResourceClient, DataClient)
  └── Discover @action methods

start_node()
  ├── Establish event_client_context
  ├── Log startup
  └── _startup() thread:
      ├── startup_handler()       ← Override this
      ├── template_handler()      ← Override for resource/location templates
      └── Status/state update loops (every N seconds)
          ├── status_handler()    ← Override to populate node_status
          └── state_handler()     ← Override to populate node_state

Per-action execution:
  ├── Parse args (TypeAnalyzer)
  ├── Validate required args
  ├── Check node ready (not busy/locked/errored)
  ├── Execute in daemon thread (with _action_lock if blocking)
  ├── Process result -> ActionResult
  └── Update action_history & log status

shutdown:
  ├── shutdown_handler()          ← Override for cleanup
  ├── Release registry identity
  └── Teardown clients

REST API Endpoints (RestNode)

Standard endpoints

Method	Path	Description
GET	/status	Node operational status
GET	/info	Node metadata, actions, capabilities
GET	/state	Custom state dict
POST	/config	Update config fields
GET	/log	EventClient logs
POST	/admin/{command}	Admin commands (lock, unlock, shutdown)
GET	/action	Action history

Per-action endpoints (3-phase file upload)

Method	Path	Description
POST	/action/{name}	Create action, get action_id
POST	/action/{name}/{id}/upload/{param}	Upload file(s) for a parameter
POST	/action/{name}/{id}/start	Start execution
GET	/action/{name}/{id}/status	Poll action status
GET	/action/{name}/{id}/result	Get result (files as key list)
GET	/action/{id}/download	Download all files as ZIP

NodeStatus

# Operational flags:
busy: bool          # Currently executing an action
paused: bool        # Manually paused
locked: bool        # Locked via admin command
stopped: bool       # Node stopped
errored: bool       # Node in error state
disconnected: bool  # Lost connection
initializing: bool  # Still starting up

# Computed:
ready: bool         # True only if no flags set and no waiting_for_config
description: str    # Human-readable reason if not ready

Resource/Location Template Registration

Declare templates as ClassVar lists for automatic registration at startup:

class MyNode(RestNode):
    resource_templates: ClassVar[list[NodeResourceTemplateDefinition]] = [
        NodeResourceTemplateDefinition(
            template_name="sample_plate",
            description="96-well sample plate",
            version="1.0",
        ),
    ]

    # Representation templates define how this node "sees" a location
    location_representation_templates: ClassVar[
        list[NodeRepresentationTemplateDefinition]
    ] = [
        NodeRepresentationTemplateDefinition(
            template_name="mynode_slot_repr",
            default_values={"slot_index": 0},
            required_overrides=["slot_index"],
            version="1.0.0",
        ),
    ]

    # Intrinsic locations are auto-created on startup with '{node_name}.' prefix
    intrinsic_locations: ClassVar[list[NodeIntrinsicLocationDefinition]] = [
        NodeIntrinsicLocationDefinition(
            location_name="slot_1",  # becomes "{node_name}.slot_1"
            representation_template_name="mynode_slot_repr",
            representation_overrides={"slot_index": 1},
            resource_template_name="sample_plate",
        ),
    ]

    def template_handler(self):
        """Called at startup to register templates. Override for custom logic."""
        super().template_handler()

Datapoint Helpers

Upload data to the Data Manager from within actions:

@action
def measure(self) -> dict:
    value = self.instrument.read()

    # Single value datapoint
    dp_id = self.create_and_upload_value_datapoint(value=value, label="measurement")

    # File datapoint
    file_id = self.create_and_upload_file_datapoint(
        file_path=Path("output.csv"), label="raw_data"
    )

    # Custom datapoint
    from madsci.common.types.data_types import DataPoint
    dp = DataPoint(label="custom", value={"nested": "data"})
    self.upload_datapoint(dp)

    return {"measurement": value, "datapoint_id": dp_id}

Testing Nodes

from fastapi.testclient import TestClient

def test_node_action():
    node = MyInstrumentNode(
        node_config=RestNodeConfig(
            enable_registry_resolution=False,  # Critical for tests
            node_url="http://127.0.0.1:2000",
        )
    )
    node.start_node(testing=True)  # Configures routes without starting uvicorn

    client = TestClient(node.rest_api)

    # Check info
    response = client.get("/info")
    assert response.status_code == 200
    assert "measure" in response.json()["actions"]

    # Run action (no file params)
    response = client.post("/action/measure", json={"sample_id": "ABC123"})
    assert response.status_code == 200
    action_id = response.json()["action_id"]

    # Start and get result
    client.post(f"/action/measure/{action_id}/start")
    import time; time.sleep(0.5)
    result = client.get(f"/action/measure/{action_id}/result")
    assert result.json()["json_result"]["sample_id"] == "ABC123"

Common Pitfalls

• ULID not UUID: Always use new_ulid_str() from madsci.common.utils for IDs
• Path not UploadFile: File parameters use pathlib.Path, not FastAPI's UploadFile
• blocking=True by default: Actions serialize unless you set blocking=False
• Registry in tests: Always set enable_registry_resolution=False in test configs
• Config discovery: NodeConfig walks up directories looking for settings.yaml, node.settings.yaml, etc. Set _settings_dir to control where it looks
• Client lazy init: self.resource_client, self.data_client, self.event_client are created on first access. If the corresponding service is unavailable, the first access will fail
• AnyUrl trailing slash: Pydantic's AnyUrl always adds a trailing slash to URLs