Drivers

The HardwareDriver protocol a native driver satisfies and the functions that register and look drivers up.

A native driver speaks a robot's wire protocol without lerobot; Robot(name, mode="real", driver="strands") builds one through the registry below. Here is the HardwareDriver protocol a driver satisfies and the functions that register and look drivers up.

Protocol

strands_robots.drivers.base.HardwareDriver

Bases: Protocol

The surface :func:~strands_robots.robot.Robot mode="real" returns.

See the module docstring for what is deliberately absent and why.

tool_name property

tool_name: str

Name the agent invokes this robot by.

tool_type property

tool_type: str

Tool kind reported to the agent runtime.

tool_spec property

tool_spec: ToolSpec

Schema describing the actions the agent may request.

stream

stream(tool_use: ToolUse, invocation_state: dict[str, Any], **kwargs: Any) -> AsyncGenerator[Any, None]

Run one agent tool call, yielding events and finally the result.

Parameters:

Name	Type	Description	Default
tool_use	ToolUse	The agent's request, carrying the tool id and parameters.	required
invocation_state	dict[str, Any]	Caller-provided state passed to the agent.	required
**kwargs	Any	Additional keyword arguments, for forward compatibility.	{}

Yields:

Type	Description
AsyncGenerator[Any, None]	Tool events, the last of which is the tool result. Spelled out
AsyncGenerator[Any, None]	rather than borrowing strands.types.tools.ToolGenerator, which
AsyncGenerator[Any, None]	is an alias for exactly this type: the reference implementation
AsyncGenerator[Any, None]	spells it out too, and naming the alias would add an SDK symbol this
AsyncGenerator[Any, None]	package does not otherwise depend on.

send_action

send_action(action: dict[str, Any], robot_name: str | None = None) -> dict[str, Any]

Command one action.

Parameters:

Name	Type	Description	Default
action	dict[str, Any]	Joint targets, keyed the way this driver names its joints.	required
robot_name	str | None	Which robot to command when the driver fronts several; None means the driver's own robot.	None

Returns:

Type	Description
dict[str, Any]	A status envelope describing what was commanded.

start_task

start_task(instruction: str, policy_port: int | None = None, policy_host: str = 'localhost', policy_provider: str = 'lerobot_local', duration: float = 30.0, **policy_kwargs: Any) -> dict[str, Any]

Start a policy-driven task in the background.

Parameters:

Name	Type	Description	Default
instruction	str	Natural-language instruction for the policy.	required
policy_port	int | None	Port the policy server listens on; None uses the provider's default.	None
policy_host	str	Host the policy server runs on.	'localhost'
policy_provider	str	Which policy provider to build.	'lerobot_local'
duration	float	Wall-clock budget for the task, in seconds.	30.0
**policy_kwargs	Any	Extra provider-specific policy options.	{}

Returns:

Type	Description
dict[str, Any]	A status envelope describing the task that started.

run_policy

run_policy(policy_object: Policy, instruction: str = '', duration: float = 30.0, n_steps: int | None = None) -> dict[str, Any]

Run an already-built policy against this robot.

Parameters:

Name	Type	Description	Default
policy_object	Policy	The policy to roll out.	required
instruction	str	Natural-language instruction handed to the policy.	''
duration	float	Wall-clock budget for the rollout, in seconds.	30.0
n_steps	int | None	Step budget; when given it wins over duration.	None

Returns:

Type	Description
dict[str, Any]	A status envelope describing the rollout.

get_task_status

get_task_status() -> dict[str, Any]

Report the running task's state.

Returns:

Type	Description
dict[str, Any]	A status envelope; the shape a caller polls between steps.

stop_task

stop_task() -> dict[str, Any]

Stop the running task.

Returns:

Type	Description
dict[str, Any]	A status envelope describing what was stopped.

get_status async

get_status() -> dict[str, Any]

Report the driver's own health and connection state.

Returns:

Type	Description
dict[str, Any]	A status envelope the mesh publishes as this peer's presence.

stop async

stop() -> None

Stop motion and any background loop, leaving the robot connected.

Annotated -> None, so it carries no verdict: a caller that needs the halt outcome reads :meth:stop_task, which decides one. That is exactly what makes the log the only place a halt this hook could not complete can be recorded, so an implementation that delegates to a halt verb must read that verb's envelope and log a non-success, naming what may still be moving. Discarding it returns from shutdown reporting a robot as stopped on the one surface that carries no way to say otherwise. :func:halt_failure_detail reads the reason out of such an envelope.

cleanup

cleanup() -> None

Release the device and every background resource held for it.

Annotated -> None, so like :meth:stop it carries no verdict, and the same obligation follows: an implementation that delegates to a halt verb must read that verb's envelope and log a non-success, naming what may still be moving. Here it is the more urgent of the two, because this hook goes on to release the channel a retry would need - a refused halt it did not report leaves a robot moving with nothing left in the process able to reach it. The release is owed either way: stopping half-way leaks the resource and leaves the robot moving.

Helpers for driver authors

The contract a mode="real" driver satisfies.

:func:~strands_robots.robot.Robot with mode="real" builds one object and hands it to everything downstream: a Strands agent invokes it as a tool, the Zenoh mesh publishes from it, and the teleop rail commands through it. Until a second implementation existed that object was always :class:strands_robots.hardware_robot.Robot, so the surface those consumers rely on was recorded nowhere - a driver author had to read a 3000-line class to learn which members are load-bearing.

:class:HardwareDriver writes that surface down. It is deliberately the measured contract rather than an aspirational one, which makes it smaller than a reader might expect:

• get_observation is not a member. A lerobot robot is a wrapper: it holds the device that owns the bus under robot, and :func:strands_robots.bus_access.read_observation takes that inner device - so the top-level driver is not the thing asked for a frame.

A native driver has no inner device, and for joint telemetry that is now resolved rather than assumed: :func:strands_robots.bus_access.joint_read_source prefers robot.robot and falls back to the driver itself, so a driver that owns its bus publishes joints on the state topic by exposing either a bus with sync_read or a get_observation, plus is_connected to say it is live. Neither is required, which is why neither is a member: a driver with no motors to report is otherwise complete. * The sensor attributes a mesh publishes (_pose, _imu, _battery and their siblings) are not members either. Every one is read with a getattr(robot, name, None) default, so a driver with no IMU publishes no IMU topic and is otherwise complete - making them optional by construction. A Protocol cannot express "optional", and requiring them would refuse a perfectly good arm for lacking a lidar.

What remains is the surface a driver must have for an agent to call it and for the mesh's command and task paths to work. The four tool members (tool_name, tool_type, tool_spec, stream) are also the abstract surface of :class:strands.tools.tools.AgentTool, which is what makes an object usable as a Strands tool at all.

Structural, not nominal: a driver satisfies this by having the members, with no import of - or inheritance from - anything here. Inheriting AgentTool is still the easy way to get the tool quarter right.

Constructor contract (a Protocol cannot express __init__): the factory builds a native driver as driver_cls(tool_name=<canonical name>, cameras=<cameras or None>, data_config=<data_config or None>, **kwargs), so a driver must accept those three keywords. Every further keyword it honours it declares as a parameter - port= among them, which stays polymorphic (a serial path, an IP address or a URL, interpreted by the driver that receives it). The signature is therefore the driver's whole keyword roster, and the factory refuses a keyword outside it (:func:constructor_keywords) rather than forwarding it into a **kwargs sink: a driver that tolerates an unread keyword reports success having ignored it, so Robot('so101', mode='real', driver='strands', prot='/dev/ttyACM0') built an arm that auto-detects a port while the caller believed they had named one. The lerobot path already refuses that typo by name; this is the mirror.

cameras is the one of the three that is not forwarded unconditionally. A driver that accepts it only for parity - which every driver shipped here does, because these robots address their cameras through their own SDK rather than through a caller-supplied config - must not be handed one it will never open: a dropped camera is invisible until a recording turns out to have no image columns. So the factory refuses a non-empty cameras= unless the class declares reads_cameras = True. Declaring it is the whole opt-in; the driver then receives the dict verbatim and owns opening, reading and closing the devices in it.

sim is the twin transport's engine. A driver that declares it is handed one by the factory on transport="twin" - built at the class's twin_keyframe when it declares one - so no driver imports the simulation package upward.

constructor_keywords

constructor_keywords(driver_cls: type) -> tuple[str, ...]

Return every keyword driver_cls's constructor binds.

Derived from the signature rather than listed anywhere, so a driver's roster cannot drift from what it reads: the keywords a driver honours are the parameters it declares (the constructor contract in this module's docstring). A **kwargs sink contributes nothing - it binds no name and reads none.

Parameters:

Name	Type	Description	Default
driver_cls	type	A driver class, typically one registered through :func:~strands_robots.drivers.register_native_driver.	required

Returns:

Type	Description
str	The parameter names, sorted. Read off the class rather than off
...	__init__, so the bound self is not one of them.

missing_driver_members

missing_driver_members(candidate: object) -> tuple[str, ...]

Report which :data:DRIVER_SURFACE members candidate does not have.

Answers for a class as well as an instance, which is what a caller holding a driver class before construction needs - :func:issubclass cannot: a Protocol declaring a @property has non-method members, and issubclass refuses those outright with TypeError.

Parameters:

Name	Type	Description	Default
candidate	object	A driver class or a built driver instance.	required

Returns:

Type	Description
str	The missing member names in sorted order; empty when candidate
...	satisfies the whole surface.

drifted_driver_parameters

drifted_driver_parameters(candidate: object) -> tuple[tuple[str, str], ...]

Report verbs whose parameters candidate spells differently from the Protocol.

:func:missing_driver_members answers whether the names on the class are all there, and that is all it can answer: it is hasattr, so a driver that renames a documented parameter satisfies it completely. A driver is invoked as an agent tool, and a dispatcher that spells the contract's own parameter names as keywords is the ordinary caller - so a renamed parameter is not a style difference, it is a TypeError raised past dispatch in place of the status envelope every verb here promises to return.

The check is the call a conforming caller makes: bind every parameter :class:HardwareDriver declares for the verb, by keyword. That admits the freedoms a driver legitimately has - extra parameters of its own, its own ordering, absorbing the ones it ignores in **kwargs - and refuses only the one thing no caller can work around, a required parameter reachable solely under a name the contract does not document.

Parameters:

Name	Type	Description	Default
candidate	object	A driver class or a built driver instance.	required

Returns:

Type	Description
tuple[str, str]	(verb, reason) pairs in sorted order, reason being the binding
...	failure; empty when every verb accepts the contract's own spelling.

declared_verbs

declared_verbs(tool_spec: ToolSpec | dict[str, Any]) -> list[str]

The action verbs a driver's own tool spec declares, in schema order.

Read back out of the spec rather than restated, so the verb list a refusal hands an agent is the one the schema really carries. A hand-copied list drifts the moment a verb is added or narrowed, and the agent then corrects itself towards a verb that is not there.

Parameters:

Name	Type	Description	Default
tool_spec	ToolSpec | dict[str, Any]	The driver's :attr:HardwareDriver.tool_spec.	required

Returns:

Type	Description
list[str]	The declared verbs, in the order the schema lists them.

undeclared_verb_error

undeclared_verb_error(driver: Any, action: Any) -> dict[str, Any]

Refuse an action the driver's own tool spec does not declare.

The enum a tool_spec carries describes the verbs an agent may send; nothing between the model and :meth:HardwareDriver.stream enforces it. A dispatcher whose last branch is a bare else therefore runs its final verb for every value the enum does not cover, and on these drivers the final verb is the write one - so a typo, a stale verb from an earlier schema, or a verb borrowed from a sibling driver halted the robot and answered status="success", leaving the caller unable to tell that its own verb had not been dispatched.

One owner rather than one per driver, because the list the refusal quotes has to come off the same schema the agent planned against (:func:declared_verbs).

Parameters:

Name	Type	Description	Default
driver	Any	The driver that was invoked; named in the refusal, and read for its :attr:HardwareDriver.tool_spec.	required
action	Any	Whatever arrived in the tool input, quoted back verbatim - it is not necessarily a string, and a caller cannot correct a value the refusal does not show.	required

Returns:

Type	Description
dict[str, Any]	A status="error" envelope naming the action and every declared verb.

policy_step

policy_step(policy_object: Any, instruction: str) -> Callable[[dict[str, Any]], Any] | None

Return the one-step callable for policy_object, or None.

:meth:HardwareDriver.run_policy types its first argument :class:~strands_robots.policies.Policy, and that class declares exactly two ways to ask for an action: :meth:~strands_robots.policies.Policy.get_actions and its synchronous wrapper :meth:~strands_robots.policies.Policy.get_actions_sync. It declares no step and no __call__, and no subclass in this package adds either - so a driver whose admission asks for step refuses every policy the package builds, while accepting objects the seam does not type. Resolving the shapes here, once, is what keeps the admission a driver performs and the call its loop makes describing the same set.

Three shapes are legitimate and all three resolve:

• a built :class:~strands_robots.policies.Policy, called as get_actions_sync(observation, instruction) - the typed contract;
• an object exposing step(observation) - the shape a control-loop policy written against a native driver already uses;
• a bare callable policy(observation).

An action chunk is unwrapped to its first action. :meth:~strands_robots.policies.Policy.get_actions returns a list whose length is the chunk horizon, but a native control loop commands one frame per step, so it needs the dict rather than the list - the same first-action convention :mod:strands_robots.hardware_robot applies when it consumes a chunk. Unwrapping for all three shapes rather than only the typed one keeps a single return contract: the returned callable answers with an action dict, or None when the policy produced no action for this step.

Parameters:

Name	Type	Description	Default
policy_object	Any	The candidate policy.	required
instruction	str	Instruction bound into the get_actions_sync call, which takes it as its second argument. Ignored by the other two shapes, neither of which accepts one.	required

Returns:

Type	Description
Callable[[dict[str, Any]], Any] | None	A callable taking an observation dict and answering with the action the
Callable[[dict[str, Any]], Any] | None	policy commanded - or None when policy_object is none of the
Callable[[dict[str, Any]], Any] | None	three shapes, which is the refusal a driver's admission renders. What
Callable[[dict[str, Any]], Any] | None	that callable answers is not narrowed to a dict: each loop validates
Callable[[dict[str, Any]], Any] | None	the action's shape against its own wire and names its own refusal, and
Callable[[dict[str, Any]], Any] | None	widening here would hide the value that refusal has to quote.

decode_motor_state

decode_motor_state(motors: Any, index: Mapping[str, int]) -> dict[str, dict[str, Any]] | None

Decode a Unitree LowState_.motor_state array into per-joint readings.

Both Unitree drivers subscribe rt/lowstate and both are handed the same fixed-length motor_state array, addressed by wire slot: the G1's unitree_hg layout declares 35 slots of which :data:~strands_robots.drivers.g1._G1_JOINT_INDEX names 29, and the Go2's unitree_go layout is read through :data:~strands_robots.drivers.go2.GO2_JOINT_INDEX's 12. The array is the only proprioception either robot publishes, so a driver that does not read it commands PD targets it cannot check against a measured pose.

One decoder rather than one per driver, for the reason the vector readers are written once: two copies of a coercion rule drift into disagreeing about which values are readings, and the copy that never gets audited is the one still manufacturing numbers.

Every field is read through getattr(motor, name, None) and coerced by :func:telemetry_float / :func:telemetry_int, so a name a firmware revision drops lands None in the record rather than a typed default. A zero here is not a harmless placeholder: q=0.0 is a valid reading of a joint at its zero position, so a defaulted read of a renamed field publishes a plausible pose - and on the G1 that pose is what a proprioceptive policy is handed at 500 Hz.

A slot the array cannot answer is skipped rather than defaulted, so a firmware carrying fewer slots than the index names costs those joints and not the rest of the frame.

Parameters:

Name	Type	Description	Default
motors	Any	The motor_state field, already defaulted to None by the caller's getattr.	required
index	Mapping[str, int]	Joint name to wire slot, as the driver's own index table declares it.	required

Returns:

Type	Description
dict[str, dict[str, Any]] | None	A mapping of joint name to a {"q", "dq", "tau_est", "temperature"}
dict[str, dict[str, Any]] | None	record for every slot the array answered, or None when the field is
dict[str, dict[str, Any]] | None	absent, bytes-like (a buffer indexes to integers, which carry none of
dict[str, dict[str, Any]] | None	these names and would read as a full-width record of None), or
dict[str, dict[str, Any]] | None	answered no slot at all. None says the array was not read, which is
dict[str, dict[str, Any]] | None	a different fact from a robot reporting no joints.

Native driver registry

Choose which driver builds a mode="real" robot, and find it.

Three questions, kept apart because they have different answers:

• Which driver? :func:resolve_driver - a name, from the caller's driver= or the robot's registry entry; otherwise the native driver when one is registered for the robot, and :data:~strands_robots.registry.DEFAULT_DRIVER (lerobot) when none is.
• Which class? :func:get_native_driver_class - the class a native driver package registered for this robot, or None.
• Which drivers could? :func:list_driver_coverage - for every registered robot, the driver names able to build it, joined from this table and the package registry's declared lerobot types. Resolution picks one of them; the join is the only surface that reports the robots for which there is nothing to pick.

The native table starts empty and is filled by :func:register_native_driver - by the drivers this package ships on import, and by any driver package. That is the seam - the point where an implementation that is not lerobot-shaped can be reached by :func:~strands_robots.robot.Robot without the factory knowing anything about it. Registering a driver is also what makes it the robot's default: a robot that declares no hardware.driver is built by its native driver once one exists, and by lerobot until then.

resolve_driver

resolve_driver(canonical: str, explicit: str | None = None) -> str

Decide which driver name builds canonical.

Precedence, highest first: the caller's explicit choice, the robot's registry hardware.driver, then :data:~strands_robots.registry.NATIVE_DRIVER when a native driver is registered for the robot, then :data:~strands_robots.registry.DEFAULT_DRIVER (lerobot) for a robot this package cannot drive itself. "auto" and None both mean "no explicit choice", so they defer to the registry and the native table.

The native driver wins over lerobot for an undeclared robot because it is the one that works out of the box: it needs no lerobot extra (torch), no lerobot calibration file, and speaks the servo bus this package maintains. A robot whose documented surface is lerobot's (the EarthRover's teleop reads) keeps lerobot by declaring hardware.driver: "lerobot" on its registry entry.

Parameters:

Name	Type	Description	Default
canonical	str	Canonical robot name (or any alias - :func:~strands_robots.registry.resolve_name is applied).	required
explicit	str | None	The caller's driver=, or None when unset.	None

Returns:

Type	Description
str	A concrete driver name - never "auto".

Raises:

Type	Description
ValueError	If explicit is not one of :data:~strands_robots.registry.DRIVER_CHOICES.

register_native_driver

register_native_driver(canonical: str, driver_cls: type, overwrite: bool = False) -> None

Register driver_cls as the native driver for canonical.

The surface is checked here rather than at build time, because here is where the mistake is made: a driver missing stream registers fine and then fails on the first agent call, one process and several minutes away from the line that is wrong.

Parameters:

Name	Type	Description	Default
canonical	str	Robot name the driver drives; resolved through :func:~strands_robots.registry.resolve_name so an alias and its canonical name cannot register two different drivers. The robot need not exist in the registry yet - a driver package may register before the entry it serves is merged, and refusing that would make registration order-dependent.	required
driver_cls	type	The class to build. Must satisfy :class:~strands_robots.drivers.base.HardwareDriver.	required
overwrite	bool	Replace an existing registration instead of refusing it.	False

Raises:

Type	Description
TypeError	If driver_cls does not satisfy the driver surface. The missing members are named - a contract violation, so the same class of error :mod:abc raises for an unimplemented abstract method.
ValueError	If canonical already has a driver and overwrite is False.

get_native_driver_class

get_native_driver_class(canonical: str) -> type | None

Return the native driver class for canonical, or None.

Parameters:

Name	Type	Description	Default
canonical	str	Robot name or alias.	required

Returns:

Type	Description
type | None	The registered class, or None when no native driver serves this
type | None	robot - which is every robot until a driver package registers one.

list_native_drivers

list_native_drivers() -> dict[str, str]

Report which robots have a native driver.

The discovery surface behind the refusal :func:~strands_robots.robot.Robot raises for driver="strands" on a robot with no native driver: a caller who asked for one needs to see what is available, not only that their choice was not.

Returns:

Type	Description
dict[str, str]	Canonical robot name -> driver class name, sorted by robot name.

list_driver_coverage

list_driver_coverage() -> dict[str, tuple[str, ...]]

Report which drivers can build each registered robot, for every robot.

:func:list_native_drivers answers half the question and :func:~strands_robots.registry.get_hardware_type the other half, and neither is a complete answer to "can I drive this robot for real". A robot may be reachable through lerobot only, through a native driver only, through both, or through neither -- and the last group is the one no existing surface reports at all, because it is defined by two absences. Assembling it by hand is how a coverage list goes stale: a robot that gained a native driver still reads as a gap until someone re-runs the join.

Reports what the two registries declare, so it needs no optional dependency and no hardware:

• "lerobot" when the robot declares a hardware.lerobot_type. That is the type string :class:~strands_robots.hardware_robot.Robot hands to lerobot's RobotConfig; whether lerobot is installed is a property of the environment, not of the robot, so it is deliberately not consulted.
• "strands" when a native driver is registered for the robot -- by this package's own :data:~strands_robots.drivers._SHIPPED_DRIVERS, or by any driver package that has called :func:register_native_driver.

Both names are :data:~strands_robots.registry.DRIVER_CHOICES values, so an entry reads as the set of driver= arguments that can build that robot. Which one wins for a robot that has both is :func:resolve_driver's answer, not this one's: coverage is about what exists, resolution about what is chosen.

Returns:

Type	Description
dict[str, tuple[str, ...]]	Canonical robot name -> the driver names that can build it, sorted by
dict[str, tuple[str, ...]]	robot name. An empty tuple means the robot is simulation-only: neither
dict[str, tuple[str, ...]]	registry can reach hardware for it, and mode="real" has nowhere to
dict[str, tuple[str, ...]]	go until a driver is written or a lerobot type is declared.

driver_choice_error

driver_choice_error(value: object, param: str, context: str) -> str | None

Report why value is not a driver name, or None if it is one.

Parameters:

Name	Type	Description	Default
value	object	The candidate driver name.	required
param	str	Parameter name to quote in the reason.	required
context	str	Calling surface to quote in the reason.	required

Returns:

Type	Description
str | None	A reason naming the accepted values, or None when value is
str | None	one of :data:~strands_robots.registry.DRIVER_CHOICES.

Shipped drivers

Driver seam for Robot(..., mode="real").

:class:~strands_robots.drivers.base.HardwareDriver is the contract a real robot satisfies; :mod:strands_robots.drivers.registry decides which implementation a given robot gets. A driver package that is not lerobot-shaped registers itself here::

from strands_robots.drivers import register_native_driver

register_native_driver("unitree_g1", G1Driver)

and Robot("unitree_g1", mode="real", driver="strands") then builds it. The drivers shipped in this package register themselves from :data:_SHIPPED_DRIVERS on import.

shipped_robot_names

shipped_robot_names(module: object, names: tuple[str, ...] | str) -> tuple[str, ...]

Resolve one table entry's robot names against the driver's own module.

Shared with the test that grades every shipped driver against the seam, so the names it checks are the names actually registered rather than a second reading of the same table.

Parameters:

Name	Type	Description	Default
module	object	The driver's imported module.	required
names	tuple[str, ...] | str	A literal tuple of canonical names, or the name of an attribute on module holding them.	required

Returns:

Type	Description
tuple[str, ...]	The canonical robot names for that entry.

Edit page