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

# Lens Events API (WebSocket)

> Real-time control and data streaming for lens sessions via WebSocket

The Lens Events API enables real-time control and data streaming for active lens sessions via WebSocket connections. This allows you to interact with Newton models, configure input streams, and receive real-time inference results.

## Connection Setup

### 1. Create a Lens Session

First, create a lens session using the REST API:

```bash theme={"system"}
curl https://api.u1.archetypeai.app/v0.5/lens/sessions/create \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{"lens_id": "lns-fd669361822b07e2-bc608aa3fdf8b4f9"}'
```

The response includes a `session_endpoint` for WebSocket connection (see [Create Lens Session](/api-reference/lens/create-session) for the full response shape):

```json theme={"system"}
{
  "session_id": "lsn-250313c2177253cbaa7a73542edd90",
  "lens_id": "lns-fd669361822b07e2-bc608aa3fdf8b4f9",
  "session_endpoint": "wss://api.u1.archetypeai.app/v0.5/lens/sessions/lsn-250313c2177253cbaa7a73542edd90",
  "session_status": "LensSessionStatus.SESSION_STATUS_REGISTERED"
}
```

### 2. Connect to WebSocket

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

  session_endpoint = "wss://api.u1.archetypeai.app/v0.5/lens/sessions/lsn-250313c2177253cbaa7a73542edd90"
  socket = websocket.create_connection(
      session_endpoint, 
      header={"Authorization": f"Bearer {API_KEY}"}
  )

  def send_and_receive_event(socket, event_data):
      socket.send_binary(json.dumps(event_data).encode())
      response = json.loads(socket.recv())
      return response
  ```

  ```javascript JavaScript theme={"system"}
  const WebSocket = require("ws");

  const socket = new WebSocket(session_endpoint, [
      "authenticationauthorization.bearer." + API_KEY, 
      "event-protocol-v1"
  ]);

  function sendAndReceiveEvent(socket, eventData) {
      return new Promise((resolve, reject) => {
          socket.send(JSON.stringify(eventData));
          socket.once('message', (data) => {
              resolve(JSON.parse(data));
          });
          socket.once('error', reject);
      });
  }
  ```
</CodeGroup>

<Note>
  For JavaScript, use the `Sec-WebSocket-Protocol` header to pass authentication and protocol version.
</Note>

## Event Message Format

All event messages are JSON objects with a required `type` field and optional `event_data` field:

```json theme={"system"}
{
  "type": "event.type",
  "event_data": {
    // Event-specific data
  }
}
```

**Response envelope:** with the single exception of `session.status` (see below), every server response carries a `type` field suffixed with `.response` and wraps the result in `event_data`:

```json theme={"system"}
{
  "type": "event.type.response",
  "event_data": {
    "is_valid": true,
    "error_messages": []
  }
}
```

When `event_data` would be empty, it is returned as `null`.

## Session Events

### session.status

Query the current status of the active session.

<CodeGroup>
  ```python Python theme={"system"}
  event_message = {"type": "session.status"}
  response = send_and_receive_event(socket, event_message)
  print(f"Session status: {response}")
  ```

  ```javascript JavaScript theme={"system"}
  const eventMessage = {"type": "session.status"};
  const response = await sendAndReceiveEvent(socket, eventMessage);
  console.log(`Session status: ${JSON.stringify(response)}`);
  ```
</CodeGroup>

**Response Example:**

Unlike other events, `session.status` returns the full session metadata record (the same shape as [Create Lens Session](/api-reference/lens/create-session)), not the `{type, event_data}` envelope.

<Warning>
  The response currently includes an `api_key` field that echoes back the caller's real API key. Treat this field as a secret — do not log, persist, or expose it in client-side code. (Server-side scrub in progress.)
</Warning>

```json theme={"system"}
{
  "session_id": "lsn-250313c2177253cbaa7a73542edd90",
  "lens_id": "lns-fd669361822b07e2-bc608aa3fdf8b4f9",
  "lens_name": "Activity Monitor",
  "session_status": "LensSessionStatus.SESSION_STATUS_RUNNING",
  "session_endpoint": "wss://api.u1.archetypeai.app/v0.5/lens/sessions/lsn-250313c2177253cbaa7a73542edd90",
  "session_ttl_sec": 86400,
  "registration_timestamp": 1779430869.7525718,
  "last_event_timestamp": 1779430869.7525718,
  "last_update_timestamp": 1779430869.9271567,
  "session_start_timestamp": 1779430869.8115742,
  "session_destroyed_timestamp": null,
  "num_active_connections": 0,
  "num_updates": 1,
  "num_inputs": 0,
  "num_outputs": 0,
  "num_events": 0,
  "input_data_stream_id": null,
  "output_data_stream_id": null,
  "lens_config": { /* full lens configuration */ }
}
```

### session.validate

Run validation checks on the current session, including health checks and model connectivity.

<CodeGroup>
  ```python Python theme={"system"}
  event_message = {"type": "session.validate"}
  response = send_and_receive_event(socket, event_message)

  if response.get("event_data", {}).get("is_valid"):
      print("Session is valid and ready")
  else:
      print(f"Validation failed: {response.get('event_data', {}).get('error_messages')}")
  ```

  ```javascript JavaScript theme={"system"}
  const eventMessage = {"type": "session.validate"};
  const response = await sendAndReceiveEvent(socket, eventMessage);

  if (response.event_data?.is_valid) {
      console.log("Session is valid and ready");
  } else {
      console.log(`Validation failed: ${response.event_data?.error_messages}`);
  }
  ```
</CodeGroup>

**Response Example:**

```json theme={"system"}
{
  "type": "session.validate.response",
  "event_data": {
    "is_valid": true,
    "error_messages": []
  }
}
```

### session.read

Read pending event messages from the platform, such as log messages or asynchronous inference results.

<Warning>
  `session.read` only returns messages for lenses whose `model_pipeline` writes to the WebSocket mailbox. Lenses that use a `server_sent_events_writer` output processor (Activity Monitor and most cookbook lenses) route inference results to the [SSE consumer endpoint](/api-reference/lens/consumer) instead — for those, `session.read` returns `event_data: null` even when [`GET /lens/sessions/metadata`](/api-reference/lens/get-sessions-metadata) shows non-zero `num_outputs`. Inspect the lens's `model_pipeline` via [`GET /lens/metadata`](/api-reference/lens/get-metadata) to confirm which output channel it uses.
</Warning>

<Info>
  Each client should generate a unique `client_id` to enable parallel reading. Use a new `client_id` to reset message stream playback.
</Info>

<CodeGroup>
  ```python Python theme={"system"}
  import uuid

  client_id = str(uuid.uuid4())[:8]  # Generate unique 8-character ID

  event_message = {
      "type": "session.read",
      "event_data": {"client_id": client_id}
  }
  response = send_and_receive_event(socket, event_message)

  messages = (response.get("event_data") or {}).get("messages", [])
  for message in messages:
      print(f"[{message['timestamp']}] {message['type']}: {message['data']}")
  ```

  ```javascript JavaScript theme={"system"}
  const clientId = Math.random().toString(16).substr(2, 8);

  const eventMessage = {
      "type": "session.read",
      "event_data": {"client_id": clientId}
  };
  const response = await sendAndReceiveEvent(socket, eventMessage);

  (response.event_data?.messages ?? []).forEach(message => {
      console.log(`[${message.timestamp}] ${message.type}: ${message.data}`);
  });
  ```
</CodeGroup>

**Response Example (no pending messages):**

```json theme={"system"}
{
  "type": "session.read.response",
  "event_data": null
}
```

**Response Example (with pending messages):**

```json theme={"system"}
{
  "type": "session.read.response",
  "event_data": {
    "messages": [
      {
        "timestamp": 1741843200.123,
        "type": "inference.result",
        "data": {
          "result": "A construction worker wearing a hard hat and safety vest"
        }
      },
      {
        "timestamp": 1741843201.456,
        "type": "log.info",
        "data": "Frame processed successfully"
      }
    ],
    "client_id": "a1b2c3d4"
  }
}
```

### session.destroy

Destroy the active session directly via the control stream.

<CodeGroup>
  ```python Python theme={"system"}
  event_message = {"type": "session.destroy"}
  response = send_and_receive_event(socket, event_message)
  print(f"Session destroyed: {response}")

  # Close WebSocket connection
  socket.close()
  ```

  ```javascript JavaScript theme={"system"}
  const eventMessage = {"type": "session.destroy"};
  const response = await sendAndReceiveEvent(socket, eventMessage);
  console.log(`Session destroyed: ${JSON.stringify(response)}`);

  // Close WebSocket connection
  socket.close();
  ```
</CodeGroup>

## Input Stream Configuration

### input\_stream.set

Configure input data streams to automatically feed data through the lens.

#### RTSP Video Stream

Connect a real-time video camera using RTSP protocol:

<CodeGroup>
  ```python Python theme={"system"}
  event_message = {
      "type": "input_stream.set",
      "event_data": {
          "stream_type": "rtsp_video_streamer",
          "stream_config": {
              "rtsp_url": "rtsp://camera.example.com:554/stream",
              "target_image_size": [360, 640],
              "target_frame_rate_hz": 0.2
          }
      }
  }
  response = send_and_receive_event(socket, event_message)
  print(f"RTSP stream configured: {response}")
  ```

  ```javascript JavaScript theme={"system"}
  const eventMessage = {
      "type": "input_stream.set",
      "event_data": {
          "stream_type": "rtsp_video_streamer",
          "stream_config": {
              "rtsp_url": "rtsp://camera.example.com:554/stream",
              "target_image_size": [360, 640],
              "target_frame_rate_hz": 0.2
          }
      }
  };
  const response = await sendAndReceiveEvent(socket, eventMessage);
  console.log(`RTSP stream configured: ${JSON.stringify(response)}`);
  ```
</CodeGroup>

#### CSV File Reader

Connect a CSV file for data streaming:

<CodeGroup>
  ```python Python theme={"system"}
  event_message = {
      "type": "input_stream.set",
      "event_data": {
          "stream_type": "csv_file_reader",
          "stream_config": {
              "file_id": "jena_climate_2009_2016.csv"
          }
      }
  }
  response = send_and_receive_event(socket, event_message)
  print(f"CSV stream configured: {response}")
  ```

  ```javascript JavaScript theme={"system"}
  const eventMessage = {
      "type": "input_stream.set",
      "event_data": {
          "stream_type": "csv_file_reader",
          "stream_config": {
              "file_id": "jena_climate_2009_2016.csv"
          }
      }
  };
  const response = await sendAndReceiveEvent(socket, eventMessage);
  ```
</CodeGroup>

### Available Input Stream Types

<AccordionGroup>
  <Accordion title="rtsp_video_streamer">
    **Description**: Real-time video streaming from RTSP cameras

    **Configuration Parameters**:

    * `rtsp_url` (string): RTSP stream URL
    * `target_image_size` (array): \[height, width] in pixels
    * `target_frame_rate_hz` (float): Frames per second to process

    **Use Cases**: Security cameras, live monitoring, real-time analysis
  </Accordion>

  <Accordion title="csv_file_reader">
    **Description**: Stream data from uploaded CSV files

    **Configuration Parameters**:

    * `file_id` (string): ID of uploaded CSV file

    **Use Cases**: Time series analysis, batch data processing, historical data analysis
  </Accordion>

  <Accordion title="video_file_reader">
    **Description**: Stream a pre-recorded video file (e.g. `.mp4`) that was previously uploaded via the [Files API](/api-reference/files/upload)

    **Configuration Parameters**:

    * `file_id` (string): `file_id` returned by the upload — typically the original filename (e.g. `"my_video.mp4"`), **not** the `file_uid` UUID

    **Use Cases**: Offline analysis of recorded footage, replaying dashcam / camera archives through a vision lens

    See [Video File Reader](/core-concepts/streams/data-connectors/readers/video-reader) for the full data-connector reference.
  </Accordion>

  <Accordion title="sensor_streamer">
    **Description**: Direct sensor data streaming

    **Configuration Parameters**:

    * Configuration varies by sensor type

    **Use Cases**: IoT sensors, accelerometers, temperature sensors
  </Accordion>
</AccordionGroup>

## Direct Model Queries

### model.query

Send direct queries to Newton models with custom data payloads.

<Note>
  `model.query` is only handled by lenses whose pipeline includes a model-query processor. Lenses that wrap a streaming processor (e.g. `lens_camera_processor`, `lens_timeseries_state_processor`) expect input via `input_stream.set` and will time out on `model.query`. Responses arrive asynchronously — long-running queries return a timeout response with `message: "Response timed out for query"` if the model has not replied within the WebSocket read window.
</Note>

#### Image QA Example

<CodeGroup>
  ```python Python theme={"system"}
  import base64

  # Load and encode image
  with open("construction_site.jpg", "rb") as img_handle:
      base64_img = base64.b64encode(img_handle.read()).decode("utf-8")

  event_message = {
      "type": "model.query",
      "event_data": {
          "model_version": "Newton::c2_4_7b_251215a172f6d7",
          "template_name": "image_qa_template_task",
          "instruction": "Answer the following question about the image:",
          "focus": "Describe the safety equipment visible in the image.",
          "max_new_tokens": 512,
          "data": [{
              "type": "base64_img",
              "base64_img": base64_img
          }],
          "sensor_metadata": {}
      }
  }

  response = send_and_receive_event(socket, event_message)
  print(f"Newton response: {response.get('event_data', {}).get('result')}")
  ```

  ```javascript JavaScript theme={"system"}
  const fs = require('fs');

  // Load and encode image
  const imageBuffer = fs.readFileSync('construction_site.jpg');
  const base64Img = imageBuffer.toString('base64');

  const eventMessage = {
      "type": "model.query",
      "event_data": {
          "model_version": "Newton::c2_4_7b_251215a172f6d7",
          "template_name": "image_qa_template_task",
          "instruction": "Answer the following question about the image:",
          "focus": "Describe the safety equipment visible in the image.",
          "max_new_tokens": 512,
          "data": [{
              "type": "base64_img",
              "base64_img": base64Img
          }],
          "sensor_metadata": {}
      }
  };

  const response = await sendAndReceiveEvent(socket, eventMessage);
  console.log(`Newton response: ${response.event_data?.result}`);
  ```
</CodeGroup>

#### Object Detection Example

<CodeGroup>
  ```python Python theme={"system"}
  event_message = {
      "type": "model.query",
      "event_data": {
          "model_version": "Newton::c2_4_7b_251215a172f6d7",
          "template_name": "image_bbox_template_task",
          "instruction": "Localize all objects in the picture and build bounding box around them.",
          "focus": "Input: [person,car,hard_hat,safety_vest]",
          "max_new_tokens": 512,
          "data": [{
              "type": "base64_img",
              "base64_img": base64_img
          }],
          "sensor_metadata": {}
      }
  }

  response = send_and_receive_event(socket, event_message)
  print(f"Detected objects: {response.get('event_data', {}).get('result')}")
  ```

  ```javascript JavaScript theme={"system"}
  const eventMessage = {
      "type": "model.query",
      "event_data": {
          "model_version": "Newton::c2_4_7b_251215a172f6d7",
          "template_name": "image_bbox_template_task",
          "instruction": "Localize all objects in the picture and build bounding box around them.",
          "focus": "Input: [person,car,hard_hat,safety_vest]",
          "max_new_tokens": 512,
          "data": [{
              "type": "base64_img",
              "base64_img": base64Img
          }],
          "sensor_metadata": {}
      }
  };

  const response = await sendAndReceiveEvent(socket, eventMessage);
  console.log(`Detected objects: ${response.event_data?.result}`);
  ```
</CodeGroup>

## Newton Model Templates

### Available Templates

<Tabs>
  <Tab title="Image QA Template">
    **Template Name**: `image_qa_template_task`

    **Purpose**: Generate narrative descriptions of images

    **Parameters**:

    * `instruction`: Question or instruction for the model
    * `focus`: Specific aspect to focus on (e.g., "safety equipment", "project status")
    * `max_new_tokens`: Maximum response length (typically 512)

    **Output**: Natural language description, wrapped in the WS response envelope.

    **Example Response**:

    ```json theme={"system"}
    {
      "type": "model.query.response",
      "event_data": {
        "result": "The image shows construction workers wearing yellow hard hats and high-visibility safety vests."
      }
    }
    ```

    **Timeout Response** (model did not reply within the read window):

    ```json theme={"system"}
    {
      "type": "model.query.response",
      "message": "Response timed out for query"
    }
    ```
  </Tab>

  <Tab title="Object Detection Template">
    **Template Name**: `image_bbox_template_task`

    **Purpose**: Detect and localize objects with bounding boxes

    **Parameters**:

    * `instruction`: Object detection instruction
    * `focus`: List of objects to detect (e.g., "Input: \[person,car,dog]")
    * `max_new_tokens`: Maximum response length

    **Output**: Structured bounding box coordinates, wrapped in the WS response envelope.

    **Example Response**:

    ```json theme={"system"}
    {
      "type": "model.query.response",
      "event_data": {
        "result": "person: [0.15, 0.20, 0.45, 0.80]\nhard_hat: [0.25, 0.15, 0.35, 0.25]\nsafety_vest: [0.20, 0.35, 0.40, 0.65]"
      }
    }
    ```
  </Tab>
</Tabs>

## Error Handling

Handle WebSocket errors and connection issues:

<CodeGroup>
  ```python Python theme={"system"}
  import websocket
  import json
  import time

  def robust_websocket_connection(session_endpoint, api_key):
      max_retries = 3
      retry_count = 0
      
      while retry_count < max_retries:
          try:
              socket = websocket.create_connection(
                  session_endpoint,
                  header={"Authorization": f"Bearer {api_key}"},
                  timeout=30
              )
              return socket
          except Exception as e:
              retry_count += 1
              print(f"Connection attempt {retry_count} failed: {e}")
              if retry_count < max_retries:
                  time.sleep(2 ** retry_count)  # Exponential backoff
              else:
                  raise e

  def safe_send_and_receive(socket, event_data, timeout=30):
      try:
          socket.settimeout(timeout)
          socket.send_binary(json.dumps(event_data).encode())
          response = json.loads(socket.recv())
          return response
      except websocket.WebSocketTimeoutException:
          raise Exception("WebSocket timeout - no response received")
      except websocket.WebSocketConnectionClosedException:
          raise Exception("WebSocket connection closed unexpectedly")
  ```

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

  function createRobustWebSocketConnection(sessionEndpoint, apiKey) {
      return new Promise((resolve, reject) => {
          const socket = new WebSocket(sessionEndpoint, [
              "authenticationauthorization.bearer." + apiKey,
              "event-protocol-v1"
          ]);
          
          const timeout = setTimeout(() => {
              socket.terminate();
              reject(new Error('WebSocket connection timeout'));
          }, 30000);
          
          socket.on('open', () => {
              clearTimeout(timeout);
              resolve(socket);
          });
          
          socket.on('error', (error) => {
              clearTimeout(timeout);
              reject(error);
          });
      });
  }

  function safeSendAndReceive(socket, eventData, timeoutMs = 30000) {
      return new Promise((resolve, reject) => {
          const timeout = setTimeout(() => {
              reject(new Error('Request timeout'));
          }, timeoutMs);
          
          socket.once('message', (data) => {
              clearTimeout(timeout);
              try {
                  resolve(JSON.parse(data));
              } catch (e) {
                  reject(new Error('Invalid JSON response'));
              }
          });
          
          socket.once('error', (error) => {
              clearTimeout(timeout);
              reject(error);
          });
          
          socket.send(JSON.stringify(eventData));
      });
  }
  ```
</CodeGroup>

## Best Practices

<AccordionGroup>
  <Accordion title="Connection Management">
    * Always authenticate WebSocket connections with your API key
    * Implement connection retry logic with exponential backoff
    * Set appropriate timeouts for requests (30 seconds recommended)
    * Close connections properly to free resources
  </Accordion>

  <Accordion title="Event Handling">
    * Use unique `client_id` for each client to enable parallel processing
    * Handle asynchronous events properly with `session.read`
    * Implement error handling for malformed responses
    * Log events for debugging and monitoring
  </Accordion>

  <Accordion title="Performance Optimization">
    * Use appropriate image sizes for video streams (360x640 recommended)
    * Set reasonable frame rates (0.2 Hz for most use cases)
    * Monitor processing times and adjust accordingly
    * Batch similar queries when possible
  </Accordion>
</AccordionGroup>

## Next Steps

<CardGroup cols={2}>
  <Card title="Examples Repository" icon="github" href="https://github.com/archetypeai/python-client/blob/main/examples">
    Explore complete working examples
  </Card>

  <Card title="Session Management" icon="gears" href="/api-reference/lens/get-sessions-metadata">
    Monitor and manage active sessions
  </Card>

  <Card title="File Upload API" icon="upload" href="/api-reference/files/upload">
    Upload files for CSV streaming
  </Card>

  <Card title="Troubleshooting" icon="wrench" href="/support/troubleshooting">
    Solve common WebSocket issues
  </Card>
</CardGroup>