Connectors

mshale-connectors v0.1.0 — adapters that pull protocol data from external lab systems and convert it to ProtocolSpec. Each connector registers via the @register decorator. 91 tests passing.

Available Connectors

Quickstart

pip install mshale-connectors

# List all registered connectors
mshale-connectors list

# Pull from Benchling
mshale-connectors pull benchling \
  --config benchling.json \
  --output ./specs/ \
  --since 2026-01-01

# Pull from Opentrons OT-2 (local directory)
mshale-connectors pull opentrons \
  --directory ./ot2-protocols/ \
  --output ./specs/

# Pull from live OT-2 robot
mshale-connectors pull opentrons \
  --robot-ip 192.168.1.42 \
  --output ./specs/

# Ingest pulled specs to mshale-api
mshale-connectors ingest ./specs/ \
  --api-base https://api.mishale.bio \
  --api-key $MSHALE_API_KEY

Opentrons OT-2 Connector

The Opentrons connector reads OT-2 protocol JSON files and maps them to ProtocolSpec. OT-2 protocols use a command-based format with commandType, liquids, pipettes, and labware top-level keys.

from mshale_connectors import ConnectorRuntime

# From a directory of .json OT-2 protocol files
runtime = ConnectorRuntime.from_config({
    "connector_id": "opentrons",
    "directory": "./ot2-protocols/",
})

records = runtime.pull()
for rec in records:
    print(rec.external_id, rec.protocol_spec.title)
    if rec.conversion_error:
        print(f"  Warning: {rec.conversion_error}")

# From a live OT-2 robot
runtime = ConnectorRuntime.from_config({
    "connector_id": "opentrons",
    "robot_ip": "192.168.1.42",
})

records = runtime.pull(since=datetime(2026, 1, 1))

Writing a Custom Connector

from mshale_connectors.base import BaseConnector, register
from mshale_connectors.models import (
    ConnectionStatus, DataSource, PulledRecord
)
from datetime import datetime

@register
class MyLabConnector(BaseConnector):
    connector_id = "my-lab"

    def __init__(self, api_url: str, api_key: str):
        self.api_url = api_url
        self.api_key = api_key

    def check_connection(self) -> ConnectionStatus:
        try:
            r = httpx.get(f"{self.api_url}/health", timeout=5)
            return ConnectionStatus.OK if r.status_code == 200 else ConnectionStatus.DEGRADED
        except Exception:
            return ConnectionStatus.UNAVAILABLE

    def list_sources(self) -> list[DataSource]:
        return [DataSource(source_id="my-lab", name="My Lab LIMS")]

    def pull(self, source_id: str, since: datetime | None = None) -> list[PulledRecord]:
        records = []
        for entry in self._fetch_entries(since):
            rec = PulledRecord(
                connector_id=self.connector_id,
                source_id=source_id,
                external_id=entry["id"],
                raw=entry,
            )
            try:
                rec.protocol_spec = self._convert(entry)
            except Exception as e:
                rec.conversion_error = str(e)
            records.append(rec)
        return records

ConnectorRuntime

The ConnectorRuntime orchestrates connector lifecycle: health checking, incremental pull (via since cursor), conversion error tracking, and retry logic.