Metadata-Version: 2.4
Name: rcp-sdk
Version: 0.1.0
Summary: RCP (Robot Context Protocol) — adapter SDK for AI-to-robot-controller communication
Author: Robot.io
License-Expression: Apache-2.0
Project-URL: Homepage, https://github.com/robot-io-ai/rcp-protocol
Project-URL: Repository, https://github.com/robot-io-ai/rcp-protocol
Project-URL: Documentation, https://github.com/robot-io-ai/rcp-protocol/blob/main/docs/rcp-specification.md
Project-URL: Issues, https://github.com/robot-io-ai/rcp-protocol/issues
Keywords: robotics,rcp,robot-context-protocol,automation,ai,industrial-robots
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Scientific/Engineering
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
License-File: NOTICE
Provides-Extra: server
Requires-Dist: websockets>=12.0; extra == "server"
Requires-Dist: zeroconf>=0.132; extra == "server"
Dynamic: license-file

# RCP SDK — Robot Context Protocol

RCP (Robot Context Protocol) is an open standard for AI-to-robot-controller communication. Think of it as MCP (Model Context Protocol), but for industrial robots. RCP defines a unified interface that lets AI assistants read robot state, validate programs, deploy code, and monitor I/O across any OEM controller — ABB, KUKA, FANUC, UR, and more — through a single adapter abstraction.

## Install

```bash
pip install rcp-sdk
```

## Quick Start

Create a new adapter by subclassing `RCPAdapter` and implementing all abstract methods:

```python
from rcp_sdk import (
    RCPAdapter,
    RobotIdentity, AdapterCapabilities, JointState, TCPPosition,
    IOState, IOSignal, ProgramInfo, ToolData,
)


class MyRobotAdapter(RCPAdapter):
    """Adapter for MyRobot controllers."""

    def __init__(self):
        self._connected = False
        self._session = None

    async def connect(self, config: dict) -> None:
        host = config["host"]
        port = config.get("port", 443)
        # Establish connection to the controller
        self._session = await my_robot_api.connect(host, port)
        self._connected = True

    async def disconnect(self) -> None:
        if self._session:
            await self._session.close()
        self._connected = False

    def is_connected(self) -> bool:
        return self._connected

    async def discover(self) -> RobotIdentity:
        info = await self._session.get_system_info()
        return RobotIdentity(
            manufacturer="MyRobot",
            model=info["model"],
            controller=info["controller"],
            firmware=info["firmware_version"],
            serial=info["serial_number"],
            joint_count=6,
            payload_kg=info.get("payload", 0),
            reach_mm=info.get("reach", 0),
        )

    def capabilities(self) -> AdapterCapabilities:
        return AdapterCapabilities(
            read=True,
            validate=True,
            deploy=True,
            execute=False,
            events=["io_changed", "program_state"],
            streaming=True,
        )

    async def get_joints(self) -> list[JointState]:
        raw = await self._session.read_joints()
        return [
            JointState(
                name=f"J{i+1}",
                position_deg=j["pos"],
                min_deg=j["min"],
                max_deg=j["max"],
                max_speed_deg_s=j["max_speed"],
            )
            for i, j in enumerate(raw)
        ]

    async def get_tcp(self) -> TCPPosition:
        pos = await self._session.read_tcp()
        return TCPPosition(
            x=pos["x"], y=pos["y"], z=pos["z"],
            rx=pos["rx"], ry=pos["ry"], rz=pos["rz"],
        )

    async def get_io(self) -> IOState:
        signals = await self._session.read_io()
        return IOState(signals=[
            IOSignal(
                name=s["name"],
                signal_type=s["type"],  # "digital_input", "digital_output", etc.
                value=s["value"],
            )
            for s in signals
        ])

    async def get_programs(self) -> list[ProgramInfo]:
        progs = await self._session.list_programs()
        return [
            ProgramInfo(
                name=p["name"],
                loaded=p.get("loaded", False),
                running=p.get("running", False),
                paused=p.get("paused", False),
            )
            for p in progs
        ]

    async def get_tools(self) -> list[ToolData]:
        tools = await self._session.read_tools()
        return [
            ToolData(
                name=t["name"],
                tcp_x=t["x"], tcp_y=t["y"], tcp_z=t["z"],
                mass_kg=t["mass"],
            )
            for t in tools
        ]
```

## Safety Model

Every RCP action has a fixed risk level. These are protocol-level constants that adapters cannot override.

| Risk Level | Gate | Actions |
|---|---|---|
| **SAFE** | None — auto-approved | `stop`, read resources, validate, capture position, diff program |
| **GUARDED** | User confirmation dialog | `upload_program`, `set_io` |
| **RESTRICTED** | User confirmation + safety interlock verification | `start_program`, `jog`, `move_to`, `home` |
| **PROHIBITED** | Protocol rejects — never implement | `override_safety`, `disable_estop`, `modify_safety_config` |

Rules:
- GUARDED and RESTRICTED actions are logged to an append-only audit trail.
- Approval tokens expire after 5 minutes.
- RESTRICTED actions call `verify_interlocks()` before execution.
- PROHIBITED actions are rejected at the protocol level. Do not implement them.

```python
from rcp_sdk import ACTION_RISK, PROHIBITED_ACTIONS

# Check risk level for an action
risk = ACTION_RISK["upload_program"]  # "GUARDED"
risk = ACTION_RISK["move_to"]         # "RESTRICTED"

# These actions are always rejected
assert "override_safety" in PROHIBITED_ACTIONS
```

## API Reference

### Abstract Methods (must implement)

| Method | Signature | Description |
|---|---|---|
| `connect` | `async def connect(self, config: dict) -> None` | Connect to the controller. Config contains `host`, `port`, credentials. |
| `disconnect` | `async def disconnect(self) -> None` | Disconnect from the controller. |
| `is_connected` | `def is_connected(self) -> bool` | Return `True` if currently connected. |
| `discover` | `async def discover(self) -> RobotIdentity` | Read the robot's identity (model, serial, firmware, joint limits). |
| `capabilities` | `def capabilities(self) -> AdapterCapabilities` | Declare what this adapter supports (read, validate, deploy, execute, events). |
| `get_joints` | `async def get_joints(self) -> list[JointState]` | Get current joint positions in degrees. |
| `get_tcp` | `async def get_tcp(self) -> TCPPosition` | Get current TCP (tool center point) position. |
| `get_io` | `async def get_io(self) -> IOState` | Get all I/O signal states. |
| `get_programs` | `async def get_programs(self) -> list[ProgramInfo]` | List programs on the controller. |
| `get_tools` | `async def get_tools(self) -> list[ToolData]` | Get configured tool definitions. |

### Optional Methods (override as needed)

| Method | Signature | Risk Level | Description |
|---|---|---|---|
| `get_program_code` | `async def get_program_code(self, module: str) -> str` | SAFE | Get source code of a program module. Returns empty string if not supported. |
| `upload_program` | `async def upload_program(self, module: str, code: str) -> UploadResult` | GUARDED | Upload a program module to the controller. |
| `capture_position` | `async def capture_position(self) -> tuple[TCPPosition, list[JointState]]` | SAFE | Capture current position. Default calls `get_tcp()` + `get_joints()`. |
| `set_io` | `async def set_io(self, signal: str, value: float) -> None` | GUARDED | Set an I/O output signal value. |
| `verify_interlocks` | `async def verify_interlocks(self) -> tuple[bool, str]` | SAFE | Check safety interlocks. Returns `(ok, reason)`. Default always passes. |
| `move_to` | `async def move_to(self, position: TCPPosition) -> dict` | RESTRICTED | Move TCP to position. Returns `{"success": bool, "message": str}`. |
| `stop` | `async def stop(self) -> None` | SAFE | Immediately stop robot motion. |
| `get_context` | `async def get_context(self) -> RobotContext` | SAFE | Aggregate all state into a context block for AI injection. Has a default implementation. |

## Type Reference

All types are Python `dataclass` instances from `rcp_sdk.models`.

### `RobotIdentity`

Describes the connected robot.

| Field | Type | Description |
|---|---|---|
| `manufacturer` | `str` | OEM name (e.g., `"ABB"`, `"KUKA"`, `"FANUC"`) |
| `model` | `str` | Robot model (e.g., `"IRB 6700"`) |
| `controller` | `str` | Controller name (e.g., `"IRC5"`, `"KRC5"`) |
| `firmware` | `str` | Firmware version string |
| `serial` | `str` | Serial number |
| `joint_count` | `int` | Number of joints (default `6`) |
| `joint_limits` | `list[JointState]` | Joint limit definitions (default empty) |
| `payload_kg` | `float` | Max payload in kg (default `0.0`) |
| `reach_mm` | `float` | Max reach in mm (default `0.0`) |

### `JointState`

A single joint's current state and limits.

| Field | Type | Description |
|---|---|---|
| `name` | `str` | Joint name (`"J1"`, `"J2"`, etc.) |
| `position_deg` | `float` | Current position in degrees |
| `min_deg` | `float` | Minimum limit in degrees (default `-360.0`) |
| `max_deg` | `float` | Maximum limit in degrees (default `360.0`) |
| `max_speed_deg_s` | `float` | Maximum speed in degrees/second (default `180.0`) |

### `TCPPosition`

Tool center point position in Cartesian space.

| Field | Type | Description |
|---|---|---|
| `x` | `float` | X position in mm |
| `y` | `float` | Y position in mm |
| `z` | `float` | Z position in mm |
| `rx` | `float` | Rotation around X — OEM-native units (see Rotation Convention below) |
| `ry` | `float` | Rotation around Y — OEM-native units |
| `rz` | `float` | Rotation around Z — OEM-native units |

### `IOSignal`

A single I/O signal.

| Field | Type | Description |
|---|---|---|
| `name` | `str` | Signal name (e.g., `"DI_01"`, `"DO_Gripper"`) |
| `signal_type` | `Literal` | One of `"digital_input"`, `"digital_output"`, `"analog_input"`, `"analog_output"` |
| `value` | `float` | Signal value — `0`/`1` for digital, `0.0`-`1.0` for analog |

### `IOState`

Container for all I/O signals.

| Field | Type | Description |
|---|---|---|
| `signals` | `list[IOSignal]` | All I/O signals on the controller |

### `ProgramInfo`

Metadata about a program on the controller.

| Field | Type | Description |
|---|---|---|
| `name` | `str` | Program/module name |
| `loaded` | `bool` | Whether the program is loaded (default `False`) |
| `running` | `bool` | Whether the program is running (default `False`) |
| `paused` | `bool` | Whether the program is paused (default `False`) |

### `ToolData`

A configured tool (end-effector) definition.

| Field | Type | Description |
|---|---|---|
| `name` | `str` | Tool name |
| `tcp_x` | `float` | TCP offset X in mm (default `0.0`) |
| `tcp_y` | `float` | TCP offset Y in mm (default `0.0`) |
| `tcp_z` | `float` | TCP offset Z in mm (default `0.0`) |
| `mass_kg` | `float` | Tool mass in kg (default `0.0`) |

### `AdapterCapabilities`

Declares what the adapter supports.

| Field | Type | Description |
|---|---|---|
| `read` | `bool` | Can read robot state (default `True`) |
| `validate` | `bool` | Can validate programs (default `False`) |
| `deploy` | `bool` | Can upload programs (default `False`) |
| `execute` | `bool` | Can execute programs / motion (default `False`) |
| `events` | `list[str]` | Supported event channels (default empty) |
| `streaming` | `bool` | Supports real-time streaming (default `False`) |

### `RobotContext`

Aggregated robot state for AI prompt injection.

| Field | Type | Description |
|---|---|---|
| `identity` | `RobotIdentity \| None` | Robot identity (default `None`) |
| `tcp` | `TCPPosition \| None` | Current TCP position (default `None`) |
| `joints` | `list[JointState]` | Current joint states (default empty) |
| `io` | `IOState` | I/O state (default empty) |
| `programs` | `list[ProgramInfo]` | Programs on controller (default empty) |
| `tools` | `list[ToolData]` | Configured tools (default empty) |

Call `context.to_prompt_block()` to get a formatted text block for LLM system prompt injection.

### `RCPEvent`

A real-time event from the controller.

| Field | Type | Description |
|---|---|---|
| `channel` | `str` | Event channel: `"io_changed"`, `"position_changed"`, `"program_state"`, `"error"`, `"cycle"`, `"safety"` |
| `data` | `dict` | Event payload (default empty) |
| `timestamp` | `float` | Unix timestamp in milliseconds (default `0.0`) |

### `UploadResult`

Result of uploading a program to the controller.

| Field | Type | Description |
|---|---|---|
| `success` | `bool` | Whether the upload succeeded |
| `message` | `str` | Human-readable result message (default empty) |
| `audit_id` | `str` | Audit trail entry ID (default empty) |

### `DiffResult`

Result of comparing local code against controller code.

| Field | Type | Description |
|---|---|---|
| `local_code` | `str` | The local version of the code |
| `controller_code` | `str` | The version currently on the controller |
| `added` | `int` | Number of lines added (default `0`) |
| `removed` | `int` | Number of lines removed (default `0`) |
| `changed` | `int` | Number of lines changed (default `0`) |

Call `diff.summary()` to get a formatted string like `"+3 -1 ~2 lines"`.

### `AuditEntry`

Append-only audit log entry for GUARDED and RESTRICTED actions.

| Field | Type | Description |
|---|---|---|
| `id` | `str` | Unique entry ID |
| `timestamp` | `str` | ISO 8601 timestamp |
| `action` | `str` | Action name (e.g., `"upload_program"`) |
| `risk_level` | `RiskLevel` | `"SAFE"`, `"GUARDED"`, `"RESTRICTED"`, or `"PROHIBITED"` |
| `approved_by` | `str` | Who approved the action |
| `robot` | `str` | Robot description (e.g., `"ABB IRB 6700 (SN: 12345)"`) |
| `params` | `dict` | Action parameters (default empty) |
| `result` | `str` | Outcome: `"success"`, `"failed"`, or `"rejected"` (default empty) |
| `diff_summary` | `str` | Diff summary for upload actions (default empty) |

### `RiskLevel`

Type alias: `Literal["SAFE", "GUARDED", "RESTRICTED", "PROHIBITED"]`

## Rotation Convention

The `rx`, `ry`, `rz` fields on `TCPPosition` hold OEM-native rotation values. Each robot manufacturer uses a different rotation representation:

| OEM | Convention | Fields |
|---|---|---|
| ABB | Quaternion (converted to Euler internally) | `rx`, `ry`, `rz` as Euler angles |
| KUKA | Euler ZYX | `rx`=A, `ry`=B, `rz`=C |
| Universal Robots | Rotation vector (axis-angle) | `rx`, `ry`, `rz` as rotation vector components |
| FANUC | W,P,R (Euler ZYZ variant) | `rx`=W, `ry`=P, `rz`=R |

Adapters are responsible for converting their OEM's native rotation format into the `rx`/`ry`/`rz` fields. Consumers of RCP data should treat these values as opaque per-OEM values unless they know the specific adapter in use.

## License

Apache 2.0 — see [LICENSE](./LICENSE) and [NOTICE](./NOTICE).

Apache 2.0 was chosen (matching the RCP specification) for its explicit patent grant — important for enterprise manufacturers adopting the protocol. Same license as Kubernetes, gRPC, Kafka, ROS 2, and TensorFlow.
