Quick Start

Overview

In this quick start guide, you’ll build a simple WiFi Controller Pilot that connects to Qaptly Desktop and executes WiFi commands. This example demonstrates the core UPP concepts in a working implementation.

Prerequisites

Qaptly Desktop version 0.15.0 or later
Python 3.8+ installed
Basic knowledge of Python and JSON

Step 1: Install Dependencies

First, install the WebSocket library:

pip install websockets

Step 2: Create Your Pilot

Create a file named wifi_pilot.py:

Important: Replace <qaptly_desktop_ip> with the actual IP address of the machine running Qaptly Desktop on your local network (e.g., 192.168.1.100). You can find this IP address in your network settings or by running ifconfig (macOS/Linux) or ipconfig (Windows).

import asyncio
import json
import uuid
from datetime import datetime
import websockets

class WiFiPilot:
    def __init__(self):
        self.pilot_id = "wifi-controller-001"
        self.websocket = None

    async def connect(self):
        """Connect to Qaptly Desktop"""
        uri = "ws://<qaptly_desktop_ip>:18765/upp"
        print(f"Connecting to {uri}...")
        self.websocket = await websockets.connect(uri)
        print("Connected!")

    async def send_message(self, msg_type, payload):
        """Send a UPP message"""
        message = {
            "id": str(uuid.uuid4()),
            "protocolVersion": "1.0.0",
            "type": msg_type,
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "payload": payload
        }
        await self.websocket.send(json.dumps(message))
        return message

    async def register(self):
        """Register this pilot with Qaptly"""
        payload = {
            "pilot": {
                "id": self.pilot_id,
                "name": "WiFi Controller",
                "version": "1.0.0",
                "vendor": "My Company",
                "description": "Controls WiFi networks for testing"
            },
            "auth": {},  # Empty for first-time connection
            "capabilities": [
                {
                    "action": "wifi.scan",
                    "name": "Scan WiFi Networks",
                    "description": "Scans for available WiFi networks",
                    "params": {
                        "type": "object",
                        "properties": {},
                        "required": []
                    },
                    "returns": {
                        "type": "object",
                        "properties": {
                            "networks": {
                                "type": "array",
                                "items": {
                                    "type": "object",
                                    "properties": {
                                        "ssid": {"type": "string"},
                                        "signal": {"type": "number"}
                                    }
                                }
                            }
                        }
                    },
                    "events": ["wifi.scan_complete"]
                },
                {
                    "action": "wifi.status",
                    "name": "Get WiFi Status",
                    "description": "Returns current WiFi connection status",
                    "params": {
                        "type": "object",
                        "properties": {},
                        "required": []
                    },
                    "returns": {
                        "type": "object",
                        "properties": {
                            "connected": {"type": "boolean"},
                            "ssid": {"type": "string"}
                        }
                    },
                    "events": []
                }
            ]
        }

        await self.send_message("register", payload)
        print("Registration sent, waiting for approval...")

    async def handle_command(self, command_id, action, params):
        """Handle a command from Qaptly"""
        print(f"Executing: {action} with params: {params}")

        if action == "wifi.scan":
            # Simulate WiFi scan
            result = {
                "networks": [
                    {"ssid": "HomeWiFi", "signal": -45},
                    {"ssid": "GuestNetwork", "signal": -72}
                ]
            }
            return {"success": True, "data": result}

        elif action == "wifi.status":
            # Simulate WiFi status
            result = {
                "connected": True,
                "ssid": "HomeWiFi"
            }
            return {"success": True, "data": result}

        else:
            return {
                "success": False,
                "error": {
                    "code": "UNKNOWN_ACTION",
                    "message": f"Unknown action: {action}"
                }
            }

    async def send_heartbeat(self):
        """Send heartbeat every 30 seconds"""
        while True:
            await asyncio.sleep(30)
            await self.send_message("heartbeat", {})
            print("❤️  Heartbeat sent")

    async def listen(self):
        """Listen for messages from Qaptly"""
        try:
            async for message in self.websocket:
                data = json.loads(message)
                msg_type = data.get("type")
                payload = data.get("payload", {})

                print(f"\n📨 Received: {msg_type}")

                if msg_type == "register_response":
                    if payload.get("success"):
                        print("✅ Registration successful!")
                        if "assignedKey" in payload:
                            print(f"🔑 Save this key: {payload['assignedKey']}")
                            print("   (Use it in 'auth.key' for future connections)")
                    elif payload.get("pending"):
                        print("⏳ Waiting for user approval in Qaptly Desktop...")
                    else:
                        error = payload.get("error", {})
                        print(f"❌ Registration failed: {error.get('message')}")

                elif msg_type == "command":
                    command_id = payload.get("commandId")
                    action = payload.get("action")
                    params = payload.get("params", {})

                    # Execute command
                    response_payload = await self.handle_command(command_id, action, params)
                    response_payload["commandId"] = command_id

                    # Send response
                    await self.send_message("response", response_payload)
                    print(f"✅ Response sent for {action}")

                elif msg_type == "subscribe":
                    events = payload.get("subscribe", [])
                    print(f"📡 Qaptly subscribed to events: {events}")

        except websockets.exceptions.ConnectionClosed:
            print("❌ Connection closed")

    async def run(self):
        """Main run loop"""
        await self.connect()
        await self.register()

        # Start heartbeat task
        heartbeat_task = asyncio.create_task(self.send_heartbeat())

        # Listen for messages
        await self.listen()

        # Cancel heartbeat when done
        heartbeat_task.cancel()

if __name__ == "__main__":
    pilot = WiFiPilot()
    asyncio.run(pilot.run())

Step 3: Start Qaptly Desktop

Launch Qaptly Desktop application
The UPP service starts automatically on port 18765

Step 4: Run Your Pilot

Execute the pilot script:

python wifi_pilot.py

You should see:

Connecting to ws://<qaptly_desktop_ip>:18765/upp...
Connected!
Registration sent, waiting for approval...

Step 5: Approve the Connection

A dialog will appear in Qaptly Desktop:

┌────────────────────────────────────────┐
│  New Pilot Connection Request         │
├────────────────────────────────────────┤
│  Name: WiFi Controller                 │
│  ID: wifi-controller-001               │
│  Vendor: My Company                    │
│                                        │
│  Capabilities:                         │
│  • wifi.scan - Scan WiFi Networks      │
│  • wifi.status - Get WiFi Status       │
│                                        │
│  [Reject]  [Approve]                   │
└────────────────────────────────────────┘

Click Approve.

Step 6: Test the Connection

Your pilot should receive:

✅ Registration successful!
🔑 Save this key: abc123-generated-key-xyz789
   (Use it in 'auth.key' for future connections)
❤️  Heartbeat sent

Step 7: Execute a Command

You can now execute commands from Qaptly. The pilot will receive commands and respond automatically. For example, if Qaptly sends a wifi.scan command, you’ll see:

📨 Received: command
Executing: wifi.scan with params: {}
✅ Response sent for wifi.scan

Next Connection (With Key)

For subsequent connections, use the assigned key:

async def register(self):
    payload = {
        "pilot": {
            "id": self.pilot_id,
            "name": "WiFi Controller",
            "version": "1.0.0"
        },
        "auth": {
            "key": "abc123-generated-key-xyz789"  # ← Use the key from first connection
        },
        "capabilities": [...]
    }
    await self.send_message("register", payload)

With a valid key, your pilot will be approved automatically without user interaction.

Key Concepts Demonstrated

This example shows: ✅ WebSocket Connection - Connect to ws://<qaptly_desktop_ip>:18765/upp ✅ Message Format - All messages follow UPP structure ✅ Registration - Declare pilot info and capabilities ✅ Authentication - First-time approval + key-based auth ✅ Command Handling - Receive and respond to commands ✅ Heartbeat - Keep connection alive every 30 seconds ✅ JSON Schema - Define params and returns for each capability

Troubleshooting

”Connection refused”

Make sure Qaptly Desktop is running
Check that no firewall blocks port 18765
Verify you’re connecting to <qaptly_desktop_ip> (localhost)

“Registration failed: AUTH_FAILED”

Your key may be invalid or expired
Try connecting without a key (first-time flow)
Check for typos in the key string

Heartbeat timeout

Your pilot must send heartbeat every 30 seconds
Connection closes after 90 seconds without any message

Next Steps

Now that you have a working pilot:

Add real WiFi control - Replace simulation with actual network commands
Implement more capabilities - Add wifi.connect, wifi.disconnect
Send events - Notify Qaptly about WiFi state changes
Read the protocol - Understand all message types and flows

Protocol Reference

Complete message specifications

Capabilities Guide

Define complex capabilities with JSON Schema

Authentication

Security model and key management

Examples

More complete implementations

Android Pilot

iOS Pilot

Chrome Extension Pilot

Custom Pilots (UPP)

Overview

Prerequisites

Step 1: Install Dependencies

Step 2: Create Your Pilot

Step 3: Start Qaptly Desktop

Step 4: Run Your Pilot

Step 5: Approve the Connection

Step 6: Test the Connection

Step 7: Execute a Command

Next Connection (With Key)

Key Concepts Demonstrated

Troubleshooting

”Connection refused”

“Registration failed: AUTH_FAILED”

Heartbeat timeout

Next Steps

Protocol Reference

Capabilities Guide

Authentication

Examples

​Overview

​Prerequisites

​Step 1: Install Dependencies

​Step 2: Create Your Pilot

​Step 3: Start Qaptly Desktop

​Step 4: Run Your Pilot

​Step 5: Approve the Connection

​Step 6: Test the Connection

​Step 7: Execute a Command

​Next Connection (With Key)

​Key Concepts Demonstrated

​Troubleshooting

​”Connection refused”

​“Registration failed: AUTH_FAILED”

​Heartbeat timeout

​Next Steps

Protocol Reference

Capabilities Guide

Authentication

Examples

Overview

Prerequisites

Step 1: Install Dependencies

Step 2: Create Your Pilot

Step 3: Start Qaptly Desktop

Step 4: Run Your Pilot

Step 5: Approve the Connection

Step 6: Test the Connection

Step 7: Execute a Command

Next Connection (With Key)

Key Concepts Demonstrated

Troubleshooting

”Connection refused”

“Registration failed: AUTH_FAILED”

Heartbeat timeout

Next Steps