Skip to content
Last updated

Connecting

Step 1: Verify Your API Key with cURL

Before writing any code, confirm your API key is valid. The -i flag shows the HTTP response headers — a 101 Switching Protocols status means authentication succeeded.

curl -i \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Connection: Upgrade" \
  -H "Upgrade: websocket" \
  -H "Sec-WebSocket-Version: 13" \
  -H "Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==" \
  wss://api4-general.collective2.com/ws

A successful response starts with:

HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade

An invalid API key returns HTTP 401 with a JSON error body.

Step 2: Open a Live Connection with websocat

Use websocat to open a live connection and see the server's first message:

websocat -H "Authorization: Bearer YOUR_API_KEY" wss://api4-general.collective2.com/ws

The server immediately sends a connection.connected message:

{
  "msg_type": "connection.connected",
  "id": "msg_a1b2c3d4e5f6a1b2c3d4e5f6a7b8",
  "timestamp": "2024-01-01T12:00:00Z",
  "v": 4
}

Your connection is ready once you receive this message.

connection.disconnected is a reserved message type for future use. The server does not currently emit it.

Step 3: Connect with Python

Install the websockets library if you haven't already:

pip install websockets
import asyncio
import json
import websockets

API_KEY = "YOUR_API_KEY"
WS_URL = "wss://api4-general.collective2.com/ws"

async def connect():
    headers = {"Authorization": f"Bearer {API_KEY}"}
    async with websockets.connect(WS_URL, additional_headers=headers) as ws:
        message = await ws.recv()
        data = json.loads(message)
        print(f"Connected: {data['msg_type']}")
        # Your logic here

asyncio.run(connect())

Heartbeat

The server sends a heartbeat message every 30 seconds to verify the connection is alive:

{
  "msg_type": "heartbeat",
  "id": "msg_a1b2c3d4e5f6a1b2c3d4e5f6a7b8",
  "timestamp": "2024-01-01T12:00:00Z",
  "v": 4
}

Your client must reply with a heartbeat.ack before the next heartbeat arrives (within 30 seconds). The server allows a 10-second grace period — 40 seconds total from the previous acknowledgement. Connections that miss a heartbeat are closed.

{
  "msg_type": "heartbeat.ack",
  "timestamp": "2024-01-01T12:00:01Z"
}

Always handle heartbeats in your client. A connection that ignores heartbeats will be closed within 40 seconds of the last acknowledgement.

Python Keep-alive Loop

import asyncio
import json
import websockets
from datetime import datetime, timezone

API_KEY = "YOUR_API_KEY"
WS_URL = "wss://api4-general.collective2.com/ws"

async def run():
    headers = {"Authorization": f"Bearer {API_KEY}"}
    async with websockets.connect(WS_URL, additional_headers=headers) as ws:
        # Wait for connection.connected
        msg = json.loads(await ws.recv())
        print(f"Connected: {msg['msg_type']}")

        async for raw in ws:
            msg = json.loads(raw)
            if msg["msg_type"] == "heartbeat":
                ack = {
                    "msg_type": "heartbeat.ack",
                    "timestamp": datetime.now(timezone.utc).isoformat()
                }
                await ws.send(json.dumps(ack))
            else:
                # Handle other messages
                print(msg)

asyncio.run(run())

Disconnecting

Close the connection cleanly when you are done. Most WebSocket libraries handle this automatically when exiting a with block.

await ws.close()