Xovis SDK

Core SDK	Integrations	Agentic Layer

An enterprise-grade integration SDK for Xovis 3D Sensors and the Xovis HUB Cloud infrastructure.

Read the Full Documentation Website →

Compliance Note: This project is an independent, open-source initiative. It is not officially affiliated with, maintained by, or endorsed by Xovis AG.

⚠️ Xovis HUB Cloud Compatibility & Rate Limits

This SDK is architected for enterprise-scale fleet orchestration. Due to the high concurrency of the HubClient and bulk_execute methods, a Xovis HUB Pro subscription is strongly suggested by the development team. Operating the SDK on the free tier may result in aggressive HTTP 429 Rate Limit exhaustion, which will disrupt automated provisioning and telemetry pipelines.

Overview

Integrating native Xovis DataPush protocols and REST APIs into enterprise data pipelines typically requires substantial boilerplate, complex state management, and strict network handling to maintain real-time DataPush ingestion (up to 12.5Hz).

This SDK abstracts the complexities of the Xovis hardware into a unified, modern, and type-safe "Universal Translator" architecture. It completely decouples raw edge telemetry from downstream infrastructure, enabling engineers to focus strictly on spatial analytics, fleet orchestration, and data warehousing.

System Data Flow

graph TB
    %% Class Definitions for High-End Professional Design
    classDef hardware fill:#1e293b,stroke:#38bdf8,stroke-width:2px,color:#f8fafc;
    classDef core fill:#0f172a,stroke:#2dd4bf,stroke-width:2px,color:#f8fafc;
    classDef agentic fill:#1e1b4b,stroke:#818cf8,stroke-width:2px,color:#f8fafc;
    classDef external fill:#0c0a09,stroke:#fb923c,stroke-width:2px,color:#f8fafc;

    subgraph "Xovis Hardware Layer"
        direction TB
        A["Physical Sensors / Spiders"]:::hardware
        H["Xovis HUB Cloud"]:::hardware
        H -- "Secure Proxy Tunnel (M2M)" --> A
    end

    subgraph "Data Plane (High Frequency Ingestion)"
        B["Xovis Ingestion Servers<br/>(TCP / UDP / HTTP Webhooks)"]:::core
        S["XovisSink<br/>(Zero-Copy Target)"]:::core
    end

    subgraph "Control & State Plane (SDK Core)"
        C["DeviceClient<br/>(Direct Local REST API)"]:::core
        F["HubClient<br/>(Cloud REST API)"]:::core
        D["Config Cache / State Buckets<br/>(3-Tier Autocomplete-Friendly)"]:::core
        
        F -- "connect_device()" --> C
        C <--> D
    end

    subgraph "Agentic & Tooling Layer"
        G["XovisAIToolkit<br/>(Type-Safe AI Tools)"]:::agentic
        M["MCP Server<br/>(Model Context Protocol)"]:::agentic
        R["Xovis CLI & Mission Control TUI"]:::agentic
    end

    %% Data Connections
    A -->|Data-Push up to 12.5Hz| B
    B -->|Sliding Buffer Extraction| S

    %% Control Connections
    A <-->|Local REST API v5| C
    H <-->|Hub REST API| F
    
    %% Fleet Integration
    D -. "Reflect State" .-> R
    F -. "Fleet Sync" .-> D

    %% AI Integration
    D --- G
    F --- G
    G --- M
    M --- LLM["LLM & Autonomous Agents<br/>(Cursor / Claude / LangChain)"]:::external

Architectural Pillars

The SDK is strictly quadrifurcated into four distinct planes to prevent blocking the asynchronous event loop during high-frequency operations while enabling autonomous systems:

1. The Data Plane (Telemetry Ingestion): A zero-copy, lock-free telemetry ingestion engine supporting high-frequency Live-Push (up to 12.5Hz) coordinates and minutely Logic-Push events over TCP, UDP, and HTTP. Read the Data Plane Documentation
2. The Control Plane (Configuration): A resilient, asynchronous HTTP engine wrapping the Xovis Edge and HUB APIs with strict Pydantic V2 schema validation and automatic Auth0 token lifecycles. Read the Control Plane Documentation
3. The Topology & State Plane (Fleet Orchestration): A memory-efficient graph engine modelling complex multisensor parent/child relations with an offline-first Native State Bucket. Read the State & Topology Documentation
4. The Agentic Layer (AI Orchestration): A Universal Tool Adapter and Model Context Protocol (MCP) server that grants autonomous orchestration capabilities to modern AI frameworks and LLMs. Read the Agentic Layer Documentation

Why this SDK?

| Logical Fleet Grouping | Managing thousands of non-stitched sensors via manual IP loops and separate authentication flows. | SDK-native DeviceGroup and HubFleetDirectory for dynamic, IDE-autosuggested bulk execution across independent networks. | | API Code Autocompletion | Parsing raw JSON/YAML payloads, navigating untyped dictionaries, and crashing on resource names containing spaces. | Regex-based dynamic namespace sanitization (dot-notation autocomplete) combined with raw name index bracket-notation fallback. | | AI Agent Orchestration | Creating unsafe custom scripts or exposing raw, unvalidated hardware APIs to autonomous LLM reasoning loops. | Safe-by-design Model Context Protocol (MCP) server and XovisAIToolkit wrapping operations in strict security guardrails. |

Progressive Coding Journey

Go from raw, high-speed physical data streaming to full stateful management and multisensor fleet orchestration in three progressive tiers:

Tier 1: Multi-Environment Discovery & 3-Tier Caching

Establish connectivity to your diverse fleet across local LAN, VPN, and secure Xovis HUB Cloud proxy tunnels. Automatically initialize the resilient 3-Tier Caching System to persist configuration states for offline autocomplete.

import asyncio
from xovis import UnifiedDeviceClient, HubClient, CachePaths

async def main():
    # Connect via HUB to manage Local, VPN, and Cloud-Proxy Tunnel connections
    async with HubClient() as hub:
        # Unified clients automatically probe and resolve different connection paths
        device_lan = UnifiedDeviceClient("192.168.1.10")
        device_vpn = UnifiedDeviceClient("10.8.0.5")
        device_hub = UnifiedDeviceClient("00:26:8c:12:34:56", hub_client=hub)
        
        # Concurrently synchronize and cache configurations for offline-first operations
        async with device_lan, device_vpn, device_hub:
            await asyncio.gather(
                device_lan.cache.sync(),
                device_vpn.cache.sync(),
                device_hub.cache.sync()
            )
            
            # Save the aggregated, autocomplete-friendly states to disk
            device_lan.cache.export_to_file(CachePaths.DEVICE_STATE)
            print(f"Successfully cached states in: {CachePaths.DEVICE_STATE}")

if __name__ == "__main__":
    asyncio.run(main())

Tier 2: Concurrent Multi-Sink Telemetry Ingestion & Port Offsetting

Prevent port collisions on multi-device networks. Load offline cache configurations via dot-notation, offset telemetry ports dynamically, spin up concurrent lock-free TCP/UDP ingestion servers, and activate live stream push agents.

import asyncio
from xovis import UnifiedDeviceClient, XovisTCPServer, XovisSink

class CustomTelemetrySink(XovisSink):
    async def write(self, frame: dict) -> None:
        # Zero-copy, high-frequency stream processing (coordinates/people tracking)
        for person in frame.get("people", []):
            print(f"Tracking ID {person['id']} at ({person['x']}, {person['y']})")

async def main():
    # Spin up two non-blocking TCP telemetry ingestion servers concurrently
    server_1 = XovisTCPServer(port=9000, sink=CustomTelemetrySink())
    server_2 = XovisTCPServer(port=10000, sink=CustomTelemetrySink())
    await asyncio.gather(server_1.start(), server_2.start())
    
    # Connect to device, mutate local ports on the connection agent to prevent conflicts
    async with UnifiedDeviceClient("10.8.0.5") as device:
        await device.cache.load_from_disk()
        
        # Fetch configurations safely via autocomplete-friendly dot-notation
        conn = device.cache.connections.by_name.MainSink
        conn.port += 1000  # Offset port dynamically
        
        # Apply the connection payload and programmatically activate the telemetry stream
        await device.datapush.update_connection(conn.id, conn)
        await device.datapush.enable_agent("TelemetryAgent")
        print("Telemetry connection offset applied and agent live stream activated!")

if __name__ == "__main__":
    asyncio.run(main())

Tier 3: Multisensor & Fleet Orchestration

Whether resolving stitched physical multisensor topologies or grouping independent sensors across your entire network into logical fleet buckets, the SDK provides concurrent broadcasting facades and offline-first IDE autosuggestions.

import asyncio
from xovis import UnifiedDeviceClient, HubClient
from xovis.api.fleet import HubFleetDirectory, DeviceGroup

async def main():
    # --- Example A: Physical Stitched Multisensor Topology ---
    async with HubClient() as hub:
        # Connect to master Spider or PC sensor to traverse the stitched multisensor topology
        async with UnifiedDeviceClient("00:26:8c:12:34:56", hub_client=hub, cache_child_devices=True) as master:
            await master.cache.sync()
            
            # Broadcast live diagnostic actions concurrently across all child sensors in one go
            context = master.cache.multisensors.by_name.Stitched_Context
            bulk_results = await context.child_devices.images.get_raw_left()
            print(f"Captured child diagnostic frames. Successes: {len(bulk_results.successes)}")

    # --- Example B: Logical Fleet Orchestration & Bulk Execution ---
    # Load the fleet directory (from Hub JSON export or live Hub API)
    directory = HubFleetDirectory.from_file()
    
    # Group independent sensors by category using offline-first IDE autosuggestions
    grp = DeviceGroup.from_directory_nodes(
        name="Office_Rollout",
        nodes=directory.by_category.Xovis_Office,
        password="SuperSecureFleetPassword123"
    )
    
    # Broadcast concurrent commands across the fleet with full IDE autosuggestions
    # Returns a resilient BulkOperationResult mapping successes and exceptions
    fleet_result = await grp.network.get_hostname()
    print(f"Successfully reached {len(fleet_result.successes)} independent devices.")

if __name__ == "__main__":
    asyncio.run(main())

Read more about the state orchestration capabilities in the State & Topology Docs

Model Context Protocol (MCP)

The Xovis SDK includes a first-class MCP server, allowing AI agents (like Claude Desktop and Cursor) to directly orchestrate hardware.

Quick Install with Smithery:

npx -y smithery install xovis-sdk

See the complete MCP Guide and AI Safety & Guardrails for more details.

Developer Experience & CLI

The SDK includes a native CLI tool to manage cache warmups, extract topology data from offline caches, generate strict Python types/Literals for perfect IDE autocompletion, alongside an interactive REPL accessor and a complete Mission Control terminal UI.

Cache Warmup & Synchronize CLI

Populate your schema definitions and local configurations directly from the terminal:

# Warm up local device config/schemas (automatically downloads OpenAPI + DataPush JSON schemas)
xovis-cli warmup 192.168.1.10 --user admin --pass secret_pass

# Warm up entire cloud fleet state and cloud schemas from Xovis HUB
xovis-cli warmup-hub --client-id MY_ID --client-secret MY_SECRET

# Generate static types from a device (local IP or MAC via Hub tunnel) for perfect autocomplete
xovis-cli generate-types --device 192.168.1.10
xovis-cli generate-types --device 00:11:22:33:44:55 --via-hub

# Launch Xovis Open SDK Mission Control TUI
xovis-cli ui

Interactive REPL Explorer

Expose your physical sensor fleets directly within interactive IPython, Jupyter Notebooks, or a standard Python REPL shell. The REPLAccessor normalized collections bypass strict validation locks for instant object-attribute discovery:

>>> from xovis import UnifiedDeviceClient
>>> device = UnifiedDeviceClient("10.8.0.5")
>>> await device.cache.load_from_disk()

# Dynamic dot-notation autocompletion with regex-sanitized names
>>> device.cache.zones.by_name.Main_Entrance
<Zone id="1" name="Main Entrance">

# Seamless list-indexing with raw names containing spaces or characters
>>> device.cache.zones["Main Entrance"]
<Zone id="1" name="Main Entrance">

Enterprise Testing & Contribution

The xovis-sdk adheres to the absolute highest tier of enterprise SDET standards, utilizing a 4-Tier test matrix with strict idempotency and hard teardown boundaries.

To contribute, please refer to our Engineering Guidelines and ensure all checks pass before submitting a PR.