API Reference

Authentication

All API requests require an API key passed as a bearer token in the Authorization header.

Authorization: Bearer YOUR_API_KEY

API keys are created and managed in the EmbodyX dashboard under Settings > API Keys. Keys are scoped to a single arm session by default. You can create an organization-scoped key for multi-arm operations from the Enterprise settings.

Base URL

The runtime API is hosted on the edge compute node at your facility. The default base URL when running locally is:

http://localhost:8747/v1

Team and Enterprise accounts with cloud management plane enabled can also use the cloud proxy endpoint shown in their dashboard.

Error codes

POST /tasks

Submit a new task for execution on the connected arm. The arm must be in an active session (see Sessions).

Request body

{
  "instruction": "Pick the red cylinder and place it in the front bin.",
  "session_id": "sess_01j2kx...",
  "options": {
    "speed_override": 0.5,
    "max_retries": 3,
    "recovery_mode": "auto"
  }
}

Response

{
  "task_id": "task_01j2kz...",
  "status": "queued",
  "created_at": "2026-03-15T09:22:14Z",
  "stream_url": "/v1/tasks/task_01j2kz.../stream"
}

GET /tasks/:id

Retrieve the current status and result of a submitted task.

Response

{
  "task_id": "task_01j2kz...",
  "status": "success",
  "instruction": "Pick the red cylinder...",
  "started_at": "2026-03-15T09:22:15Z",
  "completed_at": "2026-03-15T09:22:18Z",
  "duration_ms": 3240,
  "grasp_confidence": 0.97,
  "retries": 0
}

Possible status values: queued, executing, success, failed, cancelled.

GET /tasks/:id/stream

Server-sent events (SSE) stream for real-time task execution updates. Connect immediately after submitting a task to receive all execution events.

# Example SSE events during execution:
data: {"event": "SCENE_CAPTURE", "message": "Building scene embedding", "ts": "..."}
data: {"event": "TASK_CONDITION", "message": "Encoding instruction", "ts": "..."}
data: {"event": "ACTION_GENERATE", "message": "Trajectory at 25 Hz", "ts": "..."}
data: {"event": "ARM_EXECUTING", "message": "Arm in motion", "ts": "..."}
data: {"event": "SUCCESS", "grasp_confidence": 0.97, "ts": "..."}

POST /tasks/:id/cancel

Cancel a queued or executing task. If the arm is mid-motion, it completes the current trajectory segment and halts at a safe position before returning.

Response

{
  "task_id": "task_01j2kz...",
  "status": "cancelled",
  "cancelled_at": "2026-03-15T09:22:16Z"
}

POST /sessions

Create an arm session. A session must be active before submitting tasks. The runtime creates a default session at startup, but explicit session management is recommended for multi-arm or multi-shift deployments.

Request body

{
  "arm_id": "arm-1",
  "operator": "shift-a",
  "task_context": "Bin-picking session for M8 cylinder sort"
}

Response

{
  "session_id": "sess_01j2kx...",
  "arm_id": "arm-1",
  "status": "active",
  "created_at": "2026-03-15T08:00:00Z"
}

GET /sessions/:id

Retrieve current session status, arm connection state, and task history for this session.

POST /sessions/:id/end

End an active arm session. The arm is moved to its rest position and the adapter connection is released. Any queued tasks are cancelled before ending.

Python SDK installation

pip install embx

Requires Python 3.11+. The SDK is available on PyPI. If you are running in an air-gapped environment, download the wheel from the EmbodyX dashboard and install with pip install embx-2.x.x-py3-none-any.whl.

EmbodyXClient

from embx import EmbodyXClient

client = EmbodyXClient(
    api_key="your_api_key",
    runtime_host="localhost",  # Edge node hostname or IP
    runtime_port=8747,         # Default port
    timeout=30                 # Request timeout in seconds
)

tasks.submit()

task = client.tasks.submit(
    instruction="Pick the largest object from the bin.",
    session_id="sess_01j2kx...",  # Optional if only one session active
    speed_override=0.5,
    max_retries=3
)

print(task.task_id)    # "task_01j2kz..."
print(task.status)     # "queued"

task.stream()

for event in task.stream():
    if event.event == "SUCCESS":
        print(f"Done. Grasp confidence: {event.grasp_confidence}")
    elif event.event == "FAILED":
        print(f"Task failed: {event.error_code}")
        break
    else:
        print(event.event, event.message)

The stream terminates automatically when the task reaches a terminal state (SUCCESS, FAILED, or CANCELLED). Use task.wait() as a synchronous alternative that blocks until completion and returns the final task object.