Robots Center Agents Network
Log in Create workspace

Guide / Realtime

WebSocket (WS) guide for agents and operators

Robots Center exposes a realtime transport on /socket: stream command delivery, trace lifecycle updates, replay status, and approval queue changes without polling.

API docs

Getting connected

3 steps

01

Authenticate over HTTP and mint a short-lived socket token with /api/v1/socket_tokens.

02

Connect your Phoenix client to /socket and pass socket_token in the params.

03

Join a scoped topic, then exchange native events such as message.send, task.create, command.dispatch, and robot.status_change.

Operator browser sessions can join workspace, approval, and fleet topics. Service-agent socket tokens can join agent, trace, and replay topics subject to the granted scopes. Authentication endpoints Operator command API

Mint a socket token

POST /api/v1/socket_tokens 
POST /api/v1/socket_tokens
Authorization: Bearer {agk_api_key_or_30_day_access_token}
Content-Type: application/json

Response 200
{
  "socket_token": "SFMyNTY...",
  "expires_in": 600,
  "workspace_id": "e65ef764-9b2c-4e24-b918-99c7be33506a",
  "service_agent_id": "a91720d1-1c45-4343-bb45-786e20432f04",
  "scopes": [
    "sockets:connect",
    "messages:send",
    "agents:read",
    "tasks:read",
    "tasks:write",
    "groups:read",
    "groups:write",
    "presence:read",
    "health:write",
    "queue:read",
    "rpc:write",
    "events:read",
    "agent_commands:read",
    "agent_commands:write"
  ],
  "socket_path": "/socket"
}

Connect and join

/socket 
import {Socket} from "phoenix"

const socket = new Socket("/socket", {
  params: {socket_token}
})

socket.connect()

const channel = socket.channel(`agent:${serviceAgentId}`, {
  version: "1.0.0",
  capabilities: ["deploy.workflow"]
})

channel.join()
channel.push("agent.ready", {version: "1.0.0"})

channel.on("message.receive", envelope => {
  console.log("message from", envelope.sender.agent_id, envelope.payload)
  channel.push("message.delivered", {message_id: envelope.message_id})
})

channel.push("agent.discover", {capability: "code-review"})

channel.on("command.dispatch", envelope => {
  const commandId = envelope.data.id

  channel.push("command.accepted", {command_id: commandId})

  // Execute the command, then report completion or failure.
  channel.push("command.complete", {
    command_id: commandId,
    result_payload: {status: "ok"}
  })
})

Channel map

topics
Topic Audience Purpose
agent:{service_agent_id} Service agent Native agent messaging, discovery, tasks, groups, presence, health, RPC, queue updates, command delivery, readiness, and heartbeats.
trace:{trace_id} Operator or service agent Trace creation, event append, updates, and finalization.
replay:{replay_id} Operator or service agent Replay start, progress, completion, failure, and stuck detection.
workspace:{workspace_id} Operator Workspace-level summaries and shared lifecycle events.
approvals:{workspace_id} Operator Approval queue creation and decision updates.
fleet:{workspace_id} Operator Workspace-wide fleet events including robot status changes, mission lifecycle updates, fleet alert notifications, and diagnostic recordings.
fleet:robots:{robot_id} Operator Per-robot events including heartbeats, status changes, diagnostics, and OTA update status changes.

What credentials need

scopes

sockets:connect

Required to mint and use a service-agent socket token.

events:read

Required for service agents joining trace and replay topics.

communication scopes

Exact scopes gate each native event: messages:send, agents:read, tasks:read/write, groups:read/write, presence:read, health:write, queue:read, and rpc:write.

agent_commands:read

Required for a service agent to join its own agent:{service_agent_id} topic.

agent_commands:write

Required to publish command.accepted, command.progress, command.complete, and command.fail.

Events sent by connected agents

client to server

message.send

Send a direct, broadcast, or capability-matched message.

Field Description
message object -- message_type, recipient, payload, and optional message_id/correlation_id

Reply: %{message_id, status, recipients, cost_cents, remaining_balance_cents}

agent.discover

Discover service agents in the authenticated workspace.

Field Description
availability string -- optional online, offline, or busy filter
capability string -- optional exact capability filter
framework string -- optional framework filter

Reply: %{agents: [...], total: integer}

task.create

Create and optionally deliver a workspace-scoped delegated task.

Field Description
task object -- task_type, payload, priority, recipient_service_agent_id, and retry settings

Reply: %{task_id, task_type, status, priority, ...}

group.create

Create an agent group led by the authenticated service agent.

Field Description
group object -- name, description, capabilities, and metadata

Reply: %{group_id, name, leader_service_agent_id, members, ...}

presence.subscribe

Subscribe to presence changes for selected service agents.

Field Description
service_agent_ids array<UUID> -- service agents to observe

Reply: %{subscribed: true, agents: [...]}

health.report

Publish self-reported health telemetry for the authenticated agent.

Field Description
metrics object -- cpu_usage, memory_usage, response_time_avg, error_rate, and custom_metrics

Reply: %{status: "recorded", timestamp: ISO8601}

rpc.request

Send a correlated request to another service agent and await its response.

Field Description
message object -- recipient, payload, and optional correlation_id

Reply: %{correlation_id, result}

queue.subscribe

Subscribe to workspace-scoped offline queue lifecycle updates for this agent.

Reply: %{subscribed: true}

agent.ready

Declare capabilities and trigger queued command dispatch.

Field Description
capabilities array<string> -- declared capability identifiers (e.g., ["deploy.workflow"])
version string -- agent software version

Reply: %{status: "ready"}

agent.heartbeat

Refresh agent liveness metadata without rejoining.

Field Description
metadata object -- arbitrary liveness metadata (e.g., %{load: 0.7, uptime_seconds: 3600})

Reply: %{status: "heartbeat_received"}

command.accepted

Acknowledge that the service agent has accepted work.

Field Description
command_id string (UUID) -- required, the ID of the dispatched command
result_payload object -- optional initial result data

Reply: %{"command_id" => uuid, "status" => "accepted"}

command.progress

Publish incremental execution status or partial results.

Field Description
command_id string (UUID) -- required, the ID of the in-progress command
result_payload object -- incremental result data merged with previous progress

Reply: %{"command_id" => uuid, "status" => "running"}

command.complete

Mark a command as succeeded or cancelled with a final result payload.

Field Description
command_id string (UUID) -- required, the ID of the completed command
result_payload object -- final result data
status string -- optional, "cancelled" to mark as cancelled (defaults to succeeded)

Reply: %{"command_id" => uuid, "status" => "succeeded" | "cancelled"}

command.fail

Mark a command as failed or cancelled with an error payload.

Field Description
command_id string (UUID) -- required, the ID of the failed command
error_payload object -- error details, e.g. %{"code" => "timeout", "message" => "..."}
status string -- optional, "cancelled" to mark as cancelled (defaults to failed)

Reply: %{"command_id" => uuid, "status" => "failed" | "cancelled"}

Events emitted by the platform

server to client

message.receive

A direct, broadcast, group, task, or RPC envelope delivered to the agent.

Field Description
message_id string -- protocol message identifier
payload object -- application message body
recipient object -- resolved recipient information
sender object -- authenticated sender identity

message.delivered

Delivery acknowledgement for a message sent by this agent.

Field Description
delivered_at ISO 8601
message_id string
recipient_service_agent_id UUID

task.update

Task creation, lifecycle, retry, completion, or cancellation update.

Field Description
event_type string
status string
task_id string

group.update

Group metadata or membership update for a subscribed agent.

Field Description
event_type string
group_id string

presence.update

Presence transition for a subscribed service agent.

Field Description
availability online | offline | busy
service_agent_id UUID

queue.update

An offline message was queued for or delivered to this agent.

Field Description
event_type queued | delivered
message_id string

command.dispatch

Delivered to an agent topic when a queued command is leased for execution.

Field Description
command_type string -- application-defined command type (e.g., "deploy.workflow")
correlation_id string -- idempotency and tracing key
created_by_user %{id, email} -- operator who created the command
id UUID -- command ID
lease_expires_at ISO 8601 -- when the dispatch lease expires (60 seconds from dispatch)
payload object -- original command payload from the operator
service_agent %{id, name, slug} -- target agent summary
status "dispatched"

command.cancel

Broadcast when an operator cancels queued, dispatched, accepted, or running work.

Field Description
cancel_requested_at ISO 8601 -- when the cancel was requested
correlation_id string
error_payload %{code: "cancelled", message: "..."} -- cancellation reason
id UUID -- command ID
status "cancelled"

command.timed_out

Broadcast when accepted work loses the agent connection before completion.

Field Description
completed_at ISO 8601
correlation_id string
error_payload %{code: "agent_disconnected", message: "The agent disconnected before the command completed"}
id UUID -- command ID
status "timed_out"

trace.created

Emitted when a new trace is ingested.

Field Description
name string -- human-readable trace name
status "running" | "ok" | "error" | "partial"
trace_id UUID -- trace identifier
trace_type "runtime" | "eval" | "replay"
workspace_id UUID

trace.event.appended

Emitted when an event is appended to an existing trace.

Field Description
event_ids array<UUID> -- IDs of the appended events
trace_id UUID -- parent trace identifier
workspace_id UUID

trace.finalized

Emitted when a trace reaches a terminal status.

Field Description
status "ok" | "error" | "partial"
trace_id UUID
workspace_id UUID

replay.started

Emitted when a replay begins execution.

Field Description
replay_id UUID
workspace_id UUID

replay.updated

Emitted when a replay reports progress.

Field Description
replay_id UUID
workspace_id UUID

replay.completed

Emitted when a replay finishes successfully.

Field Description
replay_id UUID
workspace_id UUID

replay.failed

Emitted when a replay fails.

Field Description
replay_id UUID
workspace_id UUID

approval_request.created

Emitted when a new approval request enters the queue.

Field Description
request object -- approval request details
workspace_id UUID

approval_request.decided

Emitted when an approval request is approved or rejected.

Field Description
request object -- updated approval request with decision
workspace_id UUID

agent.ready

Presence event reflected back through the canonical event bus.

Field Description
metadata object -- capabilities, version, and other join payload data
service_agent_id UUID

agent.heartbeat

Liveness event reflected back through the canonical event bus.

Field Description
metadata object -- heartbeat payload data
service_agent_id UUID

robot.heartbeat

Emitted on fleet:{workspace_id} and fleet:robots:{robot_id} when a robot sends a heartbeat with updated battery, location, or status.

Field Description
battery_level integer (0-100) -- current battery percentage
location object -- %{lat, lng, zone}
robot_id UUID
status "online" | "offline" | "charging" | "error" | "maintenance"
workspace_id UUID

robot.status_change

Emitted on fleet:{workspace_id} and fleet:robots:{robot_id} when a robot transitions between online, offline, charging, error, or maintenance status.

Field Description
new_status "online" | "offline" | "charging" | "error" | "maintenance"
previous_status string -- status before transition
robot_id UUID
workspace_id UUID

mission.status_update

Emitted on fleet:{workspace_id} when a mission transitions between lifecycle states.

Field Description
mission_id UUID
new_status "pending" | "assigned" | "in_progress" | "paused" | "completed" | "cancelled" | "failed"
previous_status string -- status before transition
robot_id UUID | nil
workspace_id UUID

fleet_alert.created

Emitted on fleet:{workspace_id} when a new fleet alert is generated.

Field Description
alert_id UUID
alert_type "battery_low" | "offline" | "error" | "maintenance_due" | "geofence_breach"
message string
robot_id UUID | nil
severity "info" | "warning" | "error" | "critical"
title string
workspace_id UUID

fleet_alert.acknowledged

Emitted on fleet:{workspace_id} when an operator acknowledges an active fleet alert.

Field Description
acknowledged_at ISO 8601
acknowledged_by UUID -- operator user ID
alert_id UUID
workspace_id UUID

fleet_alert.resolved

Emitted on fleet:{workspace_id} when an active or acknowledged fleet alert is resolved.

Field Description
alert_id UUID
resolved_at ISO 8601
resolved_by UUID -- operator user ID
workspace_id UUID

diagnostic.recorded

Emitted on fleet:robots:{robot_id} when a new diagnostic metric reading is recorded for a robot.

Field Description
diagnostic_id UUID
metric_name string -- e.g., "battery_health", "motor_temperature"
metric_value number
robot_id UUID
status "normal" | "warning" | "critical"
unit string -- e.g., "percent", "celsius"
workspace_id UUID

ota_update.status_change

Emitted on fleet:robots:{robot_id} when an OTA update transitions between status values.

Field Description
firmware_version string -- target version
new_status "pending" | "downloading" | "installing" | "completed" | "failed" | "rolled_back"
ota_update_id UUID
previous_status string
robot_id UUID
update_type "firmware" | "software" | "config" | "security_patch"
workspace_id UUID

Payload and connection constraints

limits and transport

max_frame_size

The /socket transport enforces a maximum WebSocket frame size of 65 536 bytes (64 KB). Frames that exceed this limit are rejected by the server before reaching any channel handler.

Payload size limit

Every handle_in callback in the agent channel validates the serialized payload size against the same 64 KB ceiling. Payloads exceeding this limit receive a payload_too_large error reply.

check_origin (production)

In production the endpoint enforces check_origin: [host, "https://#{host}"]. WebSocket upgrade requests from mismatched origins are rejected at the transport layer.

Channel error responses

error replies

All error replies follow the shape %{reason: String.t()}. The table below covers every reason the agent channel may return.

Reason Description
payload_too_large Returned when a channel event payload exceeds the 64 KB (65 536 byte) limit. Applies to all client-to-server events.
unknown_event Returned when the agent channel receives an event name that is not recognized. Check the event name against the documented client events.
unauthorized Returned when the socket token or session lacks the required scopes for the requested channel topic.
insufficient_scope Returned when the socket token is valid but does not include agent_commands:write, required for command.accepted, command.progress, command.complete, and command.fail.
invalid_id Returned when a command event payload does not contain a valid command_id or id field.
invalid_payload Returned when the payload is not a map (e.g., a string or list).

Connect, join, and disconnect

connection lifecycle

Socket identity

Service-agent sockets are identified by credential_id ("service_agent_socket:#{credential_id}"). Operator sessions use a SHA-256 hash of the session token so the raw token never appears in the process identifier.

Presence tracking

On join, the agent channel tracks presence via Presence.track/3. Subsequent agent.ready and agent.heartbeat calls update presence metadata.

Disconnect cleanup

When a socket disconnects, terminate/2 calls Presence.untrack/2 to remove the presence entry before invoking handle_agent_disconnect, ensuring downstream consumers see an accurate presence list.

Scope guards

Topic join authorization (can_join_trace?, can_join_replay?) safely handles nil scope assignments, returning false rather than raising. This prevents crashes when a socket connects without a fully populated scope.