> ## Documentation Index
> Fetch the complete documentation index at: https://docs.allium.so/llms.txt
> Use this file to discover all available pages before exploring further.

# Websockets

> Access Allium's APIs and data streams through websockets for real-time data delivery.

Websockets provide a persistent, bidirectional connection between your application and Allium, enabling real-time data delivery with lower latency than traditional HTTP requests.

Allium offers two websocket capabilities:

1. **[APIs over Websockets](#apis-over-websockets)** - Call any Allium API through a persistent websocket connection for faster response times
2. **[Data Streams over Websockets](#data-streams-over-websockets)** - Subscribe to real-time blockchain data streams (wraps Kafka topics)

<Note>
  If you need delivery guarantees and at-least-once semantics, we recommend using [Kafka](/datastreams/kafka) or [Pub/Sub](/datastreams/pubsub) directly. Websockets are best for real-time monitoring where occasional message loss is acceptable.
</Note>

## APIs over Websockets

APIs over websockets allow you to make API requests through a persistent connection, reducing latency by eliminating the connection overhead of individual HTTP requests.

### Connection

Connect to the websocket API endpoint with your API key:

<CodeGroup>
  ```python Python theme={null}
  import websocket
  import json

  API_KEY = "your_api_key_here"
  WS_URL = "wss://api.allium.so/api/v1/developer/ws/api"

  # Connect with X-API-KEY header
  ws = websocket.create_connection(
      WS_URL,
      header={"X-API-KEY": API_KEY}
  )
  ```

  ```javascript JavaScript theme={null}
  const WebSocket = require('ws');

  const API_KEY = "your_api_key_here";
  const WS_URL = "wss://api.allium.so/api/v1/developer/ws/api";

  // Connect with X-API-KEY header
  const ws = new WebSocket(WS_URL, {
      headers: {
          "X-API-KEY": API_KEY
      }
  });
  ```

  ```go Go theme={null}
  package main

  import (
      "log"
      
      "github.com/gorilla/websocket"
  )

  func main() {
      apiKey := "your_api_key_here"
      wsURL := "wss://api.allium.so/api/v1/developer/ws/api"
      
      // Connect with X-API-KEY header
      headers := map[string][]string{
          "X-API-KEY": {apiKey},
      }
      
      conn, _, err := websocket.DefaultDialer.Dial(wsURL, headers)
      if err != nil {
          log.Fatal("dial:", err)
      }
      defer conn.Close()
  }
  ```
</CodeGroup>

### Request Format

Send requests as JSON messages with the following structure:

```json theme={null}
{
  "method": "POST",
  "id": "unique-request-id",
  "path": "/api/v1/developer/trading/hyperliquid/info/fills",
  "body": {
    "type": "userFills",
    "user": "0xD5994523d6B235862bE1790C06cFBE8547869449"
  }
}
```

**Request Fields:**

* `method` (string, required): HTTP method - `GET`, `POST`, `PUT`, `PATCH`, or `DELETE`
* `id` (string, required): Unique identifier for tracking the request/response
* `path` (string, required): API endpoint path (e.g., `/api/v1/developer/assets`)
* `body` (object, optional): Request parameters
  * For `GET` requests: URL query parameters sent as key-value pairs in `body`
  * For `POST`/`PUT`/`PATCH`/`DELETE`: Body parameters sent as key-value pairs in `body`

### Response Format

Responses are returned as JSON messages:

```json theme={null}
{
  "id": "unique-request-id",
  "status": "success",
  "data": { ... }
}
```

**Response Fields:**

* `id` (string): Matches the request ID
* `status` (string): Either `"success"` or `"error"`
* `data` (object): Response data or error message

### Example: POST Request with Body Parameters

<CodeGroup>
  ```python Python theme={null}
  import websocket
  import json
  from datetime import datetime

  API_KEY = "your_api_key_here"
  WS_URL = "wss://api.allium.so/api/v1/developer/ws/api"

  # Connect
  ws = websocket.create_connection(
      WS_URL,
      header={"X-API-KEY": API_KEY}
  )

  # Prepare POST request with body parameters
  request = {
      "method": "POST",
      "id": f"request-{datetime.now().timestamp()}",
      "path": "/api/v1/developer/trading/hyperliquid/info/fills",
      "body": {
          "type": "userFills",
          "user": "0xD5994523d6B235862bE1790C06cFBE8547869449"
      }
  }

  # Send request
  ws.send(json.dumps(request))

  # Receive response
  response = ws.recv()
  response_json = json.loads(response)

  print(response_json)

  ws.close()
  ```

  ```javascript JavaScript theme={null}
  const WebSocket = require('ws');

  const API_KEY = "your_api_key_here";
  const WS_URL = "wss://api.allium.so/api/v1/developer/ws/api";

  // Connect
  const ws = new WebSocket(WS_URL, {
      headers: {
          "X-API-KEY": API_KEY
      }
  });

  ws.on('open', () => {
      // Prepare POST request with body parameters
      const request = {
          method: "POST",
          id: `request-${Date.now()}`,
          path: "/api/v1/developer/trading/hyperliquid/info/fills",
          body: {
              type: "userFills",
              user: "0xD5994523d6B235862bE1790C06cFBE8547869449"
          }
      };
      
      // Send request
      ws.send(JSON.stringify(request));
  });

  ws.on('message', (data) => {
      const response = JSON.parse(data);
      console.log(response);
      ws.close();
  });
  ```

  ```go Go theme={null}
  package main

  import (
      "encoding/json"
      "fmt"
      "log"
      "time"
      
      "github.com/gorilla/websocket"
  )

  type Request struct {
      Method string                 `json:"method"`
      ID     string                 `json:"id"`
      Path   string                 `json:"path"`
      Body   map[string]interface{} `json:"body"`
  }

  type Response struct {
      ID     string          `json:"id"`
      Status string          `json:"status"`
      Data   json.RawMessage `json:"data"`
  }

  func main() {
      apiKey := "your_api_key_here"
      wsURL := "wss://api.allium.so/api/v1/developer/ws/api"
      
      // Connect
      headers := map[string][]string{
          "X-API-KEY": {apiKey},
      }
      
      conn, _, err := websocket.DefaultDialer.Dial(wsURL, headers)
      if err != nil {
          log.Fatal("dial:", err)
      }
      defer conn.Close()
      
      // Prepare POST request with body parameters
      request := Request{
          Method: "POST",
          ID:     fmt.Sprintf("request-%d", time.Now().Unix()),
          Path:   "/api/v1/developer/trading/hyperliquid/info/fills",
          Body: map[string]interface{}{
              "type": "userFills",
              "user": "0xD5994523d6B235862bE1790C06cFBE8547869449",
          },
      }
      
      // Send request
      err = conn.WriteJSON(request)
      if err != nil {
          log.Fatal("write:", err)
      }
      
      // Receive response
      var response Response
      err = conn.ReadJSON(&response)
      if err != nil {
          log.Fatal("read:", err)
      }
      
      fmt.Printf("%+v\n", response)
  }
  ```
</CodeGroup>

### Example: GET Request with Query Parameters

<CodeGroup>
  ```python Python theme={null}
  # GET request - parameters in body become URL query parameters
  request = {
      "method": "GET",
      "id": "get-assets-1",
      "path": "/api/v1/developer/assets",
      "body": {
          "blockchain": "ethereum",
          "limit": "10"
      }
  }

  ws.send(json.dumps(request))
  response = ws.recv()
  response_json = json.loads(response)
  ```

  ```javascript JavaScript theme={null}
  // GET request - parameters in body become URL query parameters
  const request = {
      method: "GET",
      id: "get-assets-1",
      path: "/api/v1/developer/assets",
      body: {
          blockchain: "ethereum",
          limit: "10"
      }
  };

  ws.send(JSON.stringify(request));

  ws.on('message', (data) => {
      const response = JSON.parse(data);
      console.log(response);
  });
  ```

  ```go Go theme={null}
  // GET request - parameters in body become URL query parameters
  request := Request{
      Method: "GET",
      ID:     "get-assets-1",
      Path:   "/api/v1/developer/assets",
      Body: map[string]interface{}{
          "blockchain": "ethereum",
          "limit":      "10",
      },
  }

  err = conn.WriteJSON(request)
  if err != nil {
      log.Fatal("write:", err)
  }

  var response Response
  err = conn.ReadJSON(&response)
  if err != nil {
      log.Fatal("read:", err)
  }

  fmt.Printf("%+v\n", response)
  ```
</CodeGroup>

### Concurrency & Rate Limits

* Maximum **1,000 concurrent requests** in flight per connection
* However even with the above your account rate limits still apply, so you might not be able to reach 1,000 concurrent requests in flight per connection.
* If you try to exceed this limit, you'll receive an error response:

```json theme={null}
{
  "id": "your-request-id",
  "status": "error",
  "data": "max inflight requests reached, please wait for requests to finish"
}
```

### Benefits

* **Lower latency**: No connection overhead for each request
* **Request/response matching**: Use unique IDs to track async responses

## Data Streams over Websockets

Stream real-time blockchain data directly to your application. All Kafka topics are available via websockets for easier integration.

### Connection

Connect to a data stream by specifying a topic:

<CodeGroup>
  ```python Python theme={null}
  from websocket import WebSocketApp

  API_KEY = "your_api_key_here"
  WS_URL = "wss://api.allium.so/api/v1/developer/ws/stream?topic=ethereum.blocks"

  ws = WebSocketApp(
      WS_URL,
      header={"X-API-KEY": API_KEY},
      on_open=on_open,
      on_message=on_message,
      on_error=on_error,
      on_close=on_close
  )

  ws.run_forever()
  ```

  ```javascript JavaScript theme={null}
  const WebSocket = require('ws');

  const API_KEY = "your_api_key_here";
  const WS_URL = "wss://api.allium.so/api/v1/developer/ws/stream?topic=ethereum.blocks";

  const ws = new WebSocket(WS_URL, {
      headers: {
          "X-API-KEY": API_KEY
      }
  });

  ws.on('open', onOpen);
  ws.on('message', onMessage);
  ws.on('error', onError);
  ws.on('close', onClose);
  ```

  ```go Go theme={null}
  package main

  import (
      "log"
      
      "github.com/gorilla/websocket"
  )

  func main() {
      apiKey := "your_api_key_here"
      wsURL := "wss://api.allium.so/api/v1/developer/ws/stream?topic=ethereum.blocks"
      
      headers := map[string][]string{
          "X-API-KEY": {apiKey},
      }
      
      conn, _, err := websocket.DefaultDialer.Dial(wsURL, headers)
      if err != nil {
          log.Fatal("dial:", err)
      }
      defer conn.Close()
      
      // Handle connection events
      go handleMessages(conn)
  }
  ```
</CodeGroup>

### Available Topics

All Kafka topics are available. Use the format: `{blockchain}.{data_type}`

Examples:

* `ethereum.blocks`
* `ethereum.transactions`
* `ethereum.logs`
* `base.dex_trades`
* `solana.transactions`
* `hyperliquid.fills`
* `bitcoin.blocks`

See the [Kafka documentation](/datastreams/kafka) for the complete list of available topics and schemas.

### Starting the Stream

After connecting, send a `start` message to begin receiving data:

<CodeGroup>
  ```python Python theme={null}
  def on_open(ws):
      print("Connected to WebSocket")
      
      # Start receiving messages
      start_message = {"action": "start"}
      ws.send(json.dumps(start_message))
  ```

  ```javascript JavaScript theme={null}
  function onOpen() {
      console.log("Connected to WebSocket");
      
      // Start receiving messages
      const startMessage = {action: "start"};
      ws.send(JSON.stringify(startMessage));
  }
  ```

  ```go Go theme={null}
  func onOpen(conn *websocket.Conn) {
      fmt.Println("Connected to WebSocket")
      
      // Start receiving messages
      startMessage := map[string]string{"action": "start"}
      err := conn.WriteJSON(startMessage)
      if err != nil {
          log.Println("write start:", err)
      }
  }
  ```
</CodeGroup>

### Stopping the Stream

You can pause the stream without disconnecting by sending a `stop` message:

<CodeGroup>
  ```python Python theme={null}
  # Stop receiving messages (connection remains open)
  stop_message = {"action": "stop"}
  ws.send(json.dumps(stop_message))

  # Resume later by sending start again
  start_message = {"action": "start"}
  ws.send(json.dumps(start_message))
  ```

  ```javascript JavaScript theme={null}
  // Stop receiving messages (connection remains open)
  const stopMessage = {action: "stop"};
  ws.send(JSON.stringify(stopMessage));

  // Resume later by sending start again
  const startMessage = {action: "start"};
  ws.send(JSON.stringify(startMessage));
  ```

  ```go Go theme={null}
  // Stop receiving messages (connection remains open)
  stopMessage := map[string]string{"action": "stop"}
  conn.WriteJSON(stopMessage)

  // Resume later by sending start again
  startMessage := map[string]string{"action": "start"}
  conn.WriteJSON(startMessage)
  ```
</CodeGroup>

This is useful when you need to temporarily pause data delivery, update filters, or reduce load without closing the websocket connection.

### Filtering Data (Optional)

You can filter the stream to only receive messages matching specific criteria. Send a filter with the `setFilter` action before starting the stream:

<CodeGroup>
  ```python Python theme={null}
  def on_open(ws):
      # Set a filter before starting
      filter_message = {
          "action": "setFilter",
          "data": {
              "field": "to_address",
              "operator": "exists",
              "value": True
          }
      }
      ws.send(json.dumps(filter_message))
      
      # Start the stream
      start_message = {"action": "start"}
      ws.send(json.dumps(start_message))
  ```

  ```javascript JavaScript theme={null}
  function onOpen() {
      // Set a filter before starting
      const filterMessage = {
          action: "setFilter",
          data: {
              field: "to_address",
              operator: "exists",
              value: true
          }
      };
      ws.send(JSON.stringify(filterMessage));
      
      // Start the stream
      const startMessage = {action: "start"};
      ws.send(JSON.stringify(startMessage));
  }
  ```

  ```go Go theme={null}
  func onOpen(conn *websocket.Conn) {
      // Set a filter before starting
      filterMessage := map[string]interface{}{
          "action": "setFilter",
          "data": map[string]interface{}{
              "field":    "to_address",
              "operator": "exists",
              "value":    true,
          },
      }
      conn.WriteJSON(filterMessage)
      
      // Start the stream
      startMessage := map[string]string{"action": "start"}
      conn.WriteJSON(startMessage)
  }
  ```
</CodeGroup>

The `data` field contains the filter definition. Filters support comparison operators (`=`, `!=`, `>`, `>=`, `<`, `<=`), the `in` operator for arrays, the `exists` operator, and compound AND/OR logic.

**Example with compound filter:**

<CodeGroup>
  ```python Python theme={null}
  # Filter for large Ethereum transactions
  filter_message = {
      "action": "setFilter",
      "data": {
          "op": "AND",
          "conditions": [
              {
                  "field": "blockchain",
                  "operator": "=",
                  "value": "ethereum"
              },
              {
                  "field": "value",
                  "operator": ">",
                  "value": "1000000000000000000"
              }
          ]
      }
  }
  ws.send(json.dumps(filter_message))
  ```

  ```javascript JavaScript theme={null}
  // Filter for large Ethereum transactions
  const filterMessage = {
      action: "setFilter",
      data: {
          op: "AND",
          conditions: [
              {
                  field: "blockchain",
                  operator: "=",
                  value: "ethereum"
              },
              {
                  field: "value",
                  operator: ">",
                  value: "1000000000000000000"
              }
          ]
      }
  };
  ws.send(JSON.stringify(filterMessage));
  ```

  ```go Go theme={null}
  // Filter for large Ethereum transactions
  filterMessage := map[string]interface{}{
      "action": "setFilter",
      "data": map[string]interface{}{
          "op": "AND",
          "conditions": []map[string]interface{}{
              {
                  "field":    "blockchain",
                  "operator": "=",
                  "value":    "ethereum",
              },
              {
                  "field":    "value",
                  "operator": ">",
                  "value":    "1000000000000000000",
              },
          },
      },
  }
  conn.WriteJSON(filterMessage)
  ```
</CodeGroup>

<Note>
  See the [Filter Syntax](/datastreams/filters) documentation for complete details on all operators, nested conditions, and real-world examples.
</Note>

### Receiving Messages

Messages are delivered as JSON:

<CodeGroup>
  ```python Python theme={null}
  def on_message(ws, message):
      data = json.loads(message)
      print(f"Received: {data}")
      
      # Process your blockchain data
      process_data(data)
  ```

  ```javascript JavaScript theme={null}
  function onMessage(data) {
      const message = JSON.parse(data);
      console.log("Received:", message);
      
      // Process your blockchain data
      processData(message);
  }
  ```

  ```go Go theme={null}
  func onMessage(conn *websocket.Conn) {
      var data map[string]interface{}
      err := conn.ReadJSON(&data)
      if err != nil {
          log.Println("read:", err)
          return
      }
      
      fmt.Println("Received:", data)
      
      // Process your blockchain data
      processData(data)
  }
  ```
</CodeGroup>

### Complete Example: Streaming Ethereum Blocks

<CodeGroup>
  ```python Python theme={null}
  from websocket import WebSocketApp
  import json

  API_KEY = "your_api_key_here"
  WS_URL = "wss://api.allium.so/api/v1/developer/ws/stream?topic=ethereum.blocks"

  def on_open(ws):
      print("✅ Connected to stream")
      start_message = {"action": "start"}
      ws.send(json.dumps(start_message))

  def on_message(ws, message):
      data = json.loads(message)
      block_number = data.get('number')
      print(f"New block: {block_number}")

  def on_error(ws, error):
      print(f"❌ Error: {error}")

  def on_close(ws, close_status_code, close_msg):
      print(f"🔌 Disconnected: {close_msg}")

  ws = WebSocketApp(
      WS_URL,
      header={"X-API-KEY": API_KEY},
      on_open=on_open,
      on_message=on_message,
      on_error=on_error,
      on_close=on_close
  )

  ws.run_forever()
  ```

  ```javascript JavaScript theme={null}
  const WebSocket = require('ws');

  const API_KEY = "your_api_key_here";
  const WS_URL = "wss://api.allium.so/api/v1/developer/ws/stream?topic=ethereum.blocks";

  const ws = new WebSocket(WS_URL, {
      headers: {
          "X-API-KEY": API_KEY
      }
  });

  ws.on('open', () => {
      console.log("✅ Connected to stream");
      const startMessage = {action: "start"};
      ws.send(JSON.stringify(startMessage));
  });

  ws.on('message', (message) => {
      const data = JSON.parse(message);
      const blockNumber = data.number;
      console.log(`New block: ${blockNumber}`);
  });

  ws.on('error', (error) => {
      console.log(`❌ Error: ${error}`);
  });

  ws.on('close', (code, reason) => {
      console.log(`🔌 Disconnected: ${reason}`);
  });
  ```

  ```go Go theme={null}
  package main

  import (
      "encoding/json"
      "fmt"
      "log"
      
      "github.com/gorilla/websocket"
  )

  func main() {
      apiKey := "your_api_key_here"
      wsURL := "wss://api.allium.so/api/v1/developer/ws/stream?topic=ethereum.blocks"
      
      headers := map[string][]string{
          "X-API-KEY": {apiKey},
      }
      
      conn, _, err := websocket.DefaultDialer.Dial(wsURL, headers)
      if err != nil {
          log.Fatal("dial:", err)
      }
      defer conn.Close()
      
      // Send start message
      fmt.Println("✅ Connected to stream")
      startMessage := map[string]string{"action": "start"}
      err = conn.WriteJSON(startMessage)
      if err != nil {
          log.Fatal("write start:", err)
      }
      
      // Read messages
      for {
          var data map[string]interface{}
          err := conn.ReadJSON(&data)
          if err != nil {
              log.Println("❌ Error:", err)
              break
          }
          
          blockNumber := data["number"]
          fmt.Printf("New block: %v\n", blockNumber)
      }
      
      fmt.Println("🔌 Disconnected")
  }
  ```
</CodeGroup>

## Connection Management

### Compression

Websocket compression is supported and automatically negotiated:

<CodeGroup>
  ```python Python theme={null}
  # Python websocket-client handles compression automatically
  ws = websocket.create_connection(
      WS_URL,
      header={
          "X-API-KEY": API_KEY,
          "Sec-WebSocket-Extensions": "permessage-deflate"
      }
  )
  ```

  ```javascript JavaScript theme={null}
  // Node.js ws library handles compression automatically
  const ws = new WebSocket(WS_URL, {
      headers: {
          "X-API-KEY": API_KEY,
          "Sec-WebSocket-Extensions": "permessage-deflate"
      },
      perMessageDeflate: true
  });
  ```

  ```go Go theme={null}
  // Gorilla websocket handles compression automatically
  conn, _, err := websocket.DefaultDialer.Dial(wsURL, map[string][]string{
      "X-API-KEY": {apiKey},
      "Sec-WebSocket-Extensions": {"permessage-deflate"},
  })
  ```
</CodeGroup>

### Reconnection

Implement reconnection logic for production applications:

<CodeGroup>
  ```python Python theme={null}
  import time

  def connect_with_retry(url, headers, max_retries=5):
      retries = 0
      while retries < max_retries:
          try:
              ws = websocket.create_connection(url, header=headers)
              print("Connected successfully")
              return ws
          except Exception as e:
              retries += 1
              wait_time = min(2 ** retries, 30)
              print(f"Connection failed. Retrying in {wait_time}s...")
              time.sleep(wait_time)
      
      raise Exception("Failed to connect after max retries")
  ```

  ```javascript JavaScript theme={null}
  function connectWithRetry(url, headers, maxRetries = 5) {
      let retries = 0;
      
      function connect() {
          const ws = new WebSocket(url, {headers});
          
          ws.on('open', () => {
              console.log("Connected successfully");
              retries = 0;
          });
          
          ws.on('error', (error) => {
              console.log(`Connection failed: ${error}`);
          });
          
          ws.on('close', () => {
              if (retries < maxRetries) {
                  const waitTime = Math.min(2 ** retries, 30) * 1000;
                  console.log(`Retrying in ${waitTime/1000}s...`);
                  setTimeout(connect, waitTime);
                  retries++;
              } else {
                  console.log("Failed to connect after max retries");
              }
          });
          
          return ws;
      }
      
      return connect();
  }
  ```

  ```go Go theme={null}
  func connectWithRetry(url string, headers map[string][]string, maxRetries int) (*websocket.Conn, error) {
      var conn *websocket.Conn
      var err error
      
      for retries := 0; retries < maxRetries; retries++ {
          conn, _, err = websocket.DefaultDialer.Dial(url, headers)
          if err == nil {
              fmt.Println("Connected successfully")
              return conn, nil
          }
          
          waitTime := time.Duration(math.Min(math.Pow(2, float64(retries)), 30)) * time.Second
          fmt.Printf("Connection failed. Retrying in %v...\n", waitTime)
          time.Sleep(waitTime)
      }
      
      return nil, fmt.Errorf("failed to connect after max retries")
  }
  ```
</CodeGroup>

### Error Handling

Handle common websocket errors:

<CodeGroup>
  ```python Python theme={null}
  try:
      ws.send(json.dumps(request))
      response = ws.recv()
  except websocket.WebSocketTimeoutException:
      print("Request timeout")
  except websocket.WebSocketConnectionClosedException:
      print("Connection closed unexpectedly")
      # Implement reconnection
  except Exception as e:
      print(f"Error: {e}")
  ```

  ```javascript JavaScript theme={null}
  ws.on('error', (error) => {
      console.log(`Error: ${error}`);
      // Implement reconnection logic
  });

  ws.on('close', (code, reason) => {
      if (code !== 1000) {
          console.log(`Connection closed unexpectedly: ${code} - ${reason}`);
          // Implement reconnection logic
      }
  });

  // Timeout handling
  const timeout = setTimeout(() => {
      console.log("Request timeout");
      ws.close();
  }, 30000);

  ws.on('message', (data) => {
      clearTimeout(timeout);
      // Process message
  });
  ```

  ```go Go theme={null}
  // Read with timeout
  conn.SetReadDeadline(time.Now().Add(30 * time.Second))

  var response Response
  err := conn.ReadJSON(&response)
  if err != nil {
      if websocket.IsCloseError(err, websocket.CloseNormalClosure) {
          log.Println("Connection closed normally")
      } else if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
          log.Println("Request timeout")
      } else {
          log.Println("Connection closed unexpectedly:", err)
          // Implement reconnection logic
      }
  }
  ```
</CodeGroup>

## Authentication

Both websocket types require an API key passed in the connection headers:

<CodeGroup>
  ```python Python theme={null}
  header = {"X-API-KEY": "your_api_key_here"}
  ```

  ```javascript JavaScript theme={null}
  const headers = {
      "X-API-KEY": "your_api_key_here"
  };
  ```

  ```go Go theme={null}
  headers := map[string][]string{
      "X-API-KEY": {"your_api_key_here"},
  }
  ```
</CodeGroup>

* **APIs over Websockets**: Requires it to be enabled on your account
* **Streams over Websockets**: Requires it to be enabled and topic access

Contact account representative to enable websocket features on your account.

## Next Steps

<CardGroup cols={3}>
  <Card title="Filter Syntax" icon="filter" href="/datastreams/filters">
    Learn how to filter streams with complex conditions
  </Card>

  <Card title="Kafka Streams" icon="layer-group" href="/datastreams/kafka">
    Production-grade streaming with delivery guarantees
  </Card>

  <Card title="API Reference" icon="book" href="/api/developer/overview">
    Explore available API endpoints
  </Card>
</CardGroup>