An open protocol for AI agents and devices.
Device Connect is the open-source protocol that lets AI agents safely operate physical devices. Connect a robot, sensor, or instrument once, and every agent across every framework can find and use it.
See it → View on GitHubA standard protocol
between agents and devices.
Connecting agents to the physical world has meant a one-off integration for every fleet, every framework, every protocol. Device Connect collapses that into a device driver on the hardware and one import on the agent. The agent treats devices the way it treats any other tool. Discover, call, listen.
Four tools for your agent.
The SDK exposes everything as four MCP-compatible tools. Anything you can do with a device (query state, run a command, react to an event) goes through one of them.
Works with the frameworks you already use.
'/><use href='%23a' transform='rotate(120 1203 1203)'/><use href='%23a' transform='rotate(180 1203 1203)'/><use href='%23a' transform='rotate(240 1203 1203)'/><use href='%23a' transform='rotate(300 1203 1203)'/></svg>)
Three packages.
Device Connect ships as three pip packages. One runs on devices, one runs in agents, and the third (optional) runs on a server. They communicate over a network with three messaging options.
@rpc, @emit, @periodic, @on. The runtime handles transport, identity, heartbeats, and discovery.A device. An agent.
The whole story end-to-end: a temperature sensor that exposes a reading and emits alerts, and an agent that finds it, calls it, and listens for trouble.
# sensor.py: runs on the device from device_connect_edge import DeviceDriver, rpc, emit, periodic class TempSensor(DeviceDriver): device_type = "sensor" device_id = "temp-bay-01" @rpc() async def get_reading(self) -> dict: """Return current temperature.""" return {"temp": 22.5, "humidity": 45} @emit() async def alert(self, level, message): """Broadcast an alert event.""" pass @periodic(interval=10) async def poll(self): reading = await self.get_reading() if reading["temp"] > 30: await self.alert("warning", "Hot!")
# agent.py: runs anywhere from device_connect_agent_tools import ( connect, discover_devices, invoke_device, subscribe, ) connect() # Find what's on the network devices = discover_devices(device_type="sensor") # Call the @rpc function remotely result = invoke_device("sensor-001", "get_reading") # → {"temp": 22.5, "humidity": 45} # Listen for @emit events subscribe("*.event.alert", on_alert)
Scale from local to a production fleet.
Your code stays the same either way. To run in production, add the optional device-connect-server package and turn on security.
- Setuppip install, no servers, no certs
- MeshZenoh with LAN multicast discovery
- Scale~50–100 devices on a local network
- StateEphemeral · lives in-memory
- SecurityApp-layer auth optional
- SetupDocker compose · etcd, registry, security service
- MeshNATS with JWT/NKey identity, cloud routing
- ScaleThousands of devices across sites
- StatePersistent · etcd-backed
- SecuritymTLS, ACLs, signed commissioning, full audit log
From pip install to first RPC, in minutes.
Then write a class that subclasses Device, decorate a method with @rpc, and run it. Discovery is automatic on the LAN.
Import connect, discover_devices, invoke_device, subscribe. Pass the tool surface to your agent runtime.
For detailed guidance, see Device Connect Learning Paths:
