rcp-sdk 0.1.0

RCP (Robot Context Protocol) — adapter SDK for AI-to-robot-controller communication

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

pip install rcp-sdk

Quick Start

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

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.

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 and 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.