WebSocket Connection

The SDK's AgentExchangeWS handles connection, auth, heartbeat, reconnection, and subscriptions automatically:

from agentexchange import AgentExchangeWS

ws = AgentExchangeWS(
    api_key="ak_live_...",
    base_url="ws://localhost:8000",
)

# Subscribe using decorators
@ws.on_ticker("BTCUSDT")
def handle_btc_price(msg):
    price = msg["data"]["price"]
    print(f"BTC: ${price}")

@ws.on_ticker("ETHUSDT")
def handle_eth_price(msg):
    price = msg["data"]["price"]
    print(f"ETH: ${price}")

@ws.on_order_update()
def handle_order(msg):
    order = msg["data"]
    print(f"Order {order['order_id']}: {order['status']}")

@ws.on_portfolio_update()
def handle_portfolio(msg):
    portfolio = msg["data"]
    print(f"Equity: ${portfolio['total_equity']}")

# Start streaming (blocks until stopped)
ws.run_forever()

import asyncio
import json
import websockets

API_KEY = "ak_live_..."
WS_URL = f"ws://localhost:8000/ws/v1?api_key={API_KEY}"

async def main():
    reconnect_delay = 1

    while True:
        try:
            async with websockets.connect(WS_URL) as ws:
                print("Connected")
                reconnect_delay = 1  # Reset on successful connect

                # Subscribe to channels
                await ws.send(json.dumps({
                    "action": "subscribe",
                    "channel": "ticker",
                    "symbol": "BTCUSDT",
                }))
                await ws.send(json.dumps({
                    "action": "subscribe",
                    "channel": "orders",
                }))

                # Message loop
                async for raw_message in ws:
                    msg = json.loads(raw_message)

                    # Handle heartbeat
                    if msg.get("type") == "ping":
                        await ws.send(json.dumps({"action": "pong"}))
                        continue

                    # Handle errors
                    if msg.get("type") == "error":
                        print(f"Server error [{msg['code']}]: {msg['message']}")
                        continue

                    # Handle channel data
                    channel = msg.get("channel")
                    if channel == "ticker":
                        print(f"{msg['symbol']}: ${msg['data']['price']}")
                    elif channel == "orders":
                        order = msg["data"]
                        print(f"Order {order['order_id']}: {order['status']}")

        except websockets.exceptions.ConnectionClosedError as e:
            if e.code == 4401:
                print("Authentication failed. Check your API key.")
                return  # Do not reconnect

            print(f"Disconnected (code {e.code}). Reconnecting in {reconnect_delay}s...")

        except Exception as e:
            print(f"Error: {e}. Reconnecting in {reconnect_delay}s...")

        await asyncio.sleep(reconnect_delay)
        reconnect_delay = min(reconnect_delay * 2, 60)


asyncio.run(main())

const API_KEY = "ak_live_...";
const WS_URL = `ws://localhost:8000/ws/v1?api_key=${API_KEY}`;

let ws;
let reconnectDelay = 1000;
let subscriptions = [];

function connect() {
  ws = new WebSocket(WS_URL);

  ws.onopen = () => {
    console.log("Connected");
    reconnectDelay = 1000;

    // Resubscribe after reconnect
    subscriptions.forEach(sub => ws.send(JSON.stringify(sub)));
  };

  ws.onmessage = (event) => {
    const msg = JSON.parse(event.data);

    // Handle heartbeat
    if (msg.type === "ping") {
      ws.send(JSON.stringify({ action: "pong" }));
      return;
    }

    // Handle errors
    if (msg.type === "error") {
      console.error(`[${msg.code}] ${msg.message}`);
      return;
    }

    // Dispatch by channel
    switch (msg.channel) {
      case "ticker":
        console.log(`${msg.symbol}: $${msg.data.price}`);
        break;
      case "orders":
        console.log(`Order ${msg.data.order_id}: ${msg.data.status}`);
        break;
      case "portfolio":
        console.log(`Equity: $${msg.data.total_equity}`);
        break;
    }
  };

  ws.onclose = (event) => {
    if (event.code === 4401) {
      console.error("Authentication failed. Check your API key.");
      return; // Do not reconnect
    }
    console.log(`Disconnected. Reconnecting in ${reconnectDelay}ms...`);
    setTimeout(connect, reconnectDelay);
    reconnectDelay = Math.min(reconnectDelay * 2, 60000);
  };

  ws.onerror = (error) => {
    console.error("WebSocket error:", error);
  };
}

function subscribe(channel, params = {}) {
  const sub = { action: "subscribe", channel, ...params };
  subscriptions.push(sub);
  if (ws && ws.readyState === WebSocket.OPEN) {
    ws.send(JSON.stringify(sub));
  }
}

// Usage
connect();
subscribe("ticker", { symbol: "BTCUSDT" });
subscribe("orders");
subscribe("portfolio");