Filigran SSEClient

Common errors

• ModuleNotFoundError: No module named 'filigran_sseclient'

  cause The `filigran-sseclient` package is not installed or the Python environment is incorrect.
  fix Run `pip install filigran-sseclient` in your active Python environment.

• TypeError: SSEClient.__init__() got an unexpected keyword argument 'token'

  cause The constructor for `SSEClient` expects `api_key` for authentication, not `token`.
  fix Change `token=your_token` to `api_key=your_api_key` when initializing `SSEClient`.

• requests.exceptions.ConnectionError: HTTPSConnectionPool(host='your.opencti.url', port=443): Max retries exceeded with url: /sse (Caused by NewConnectionError('<urllib3.connection.HTTPSConnection object at 0x...>: Failed to establish a new connection: [Errno 111] Connection refused'))

  cause The client could not connect to the specified `base_url`. This could be due to an incorrect URL, firewall issues, the OpenCTI server being down, or incorrect proxy settings.
  fix Verify that `OPENCTI_BASE_URL` is correct and accessible. Check network connectivity to the OpenCTI instance. If using proxies, ensure they are configured correctly for the `requests` library (e.g., via `http_proxy`/`https_proxy` environment variables).

Warnings

• gotcha The client's `events()` method yields raw `Event` objects. The `event.data` attribute is a string that typically contains JSON. Users must manually parse this data (e.g., `json.loads(event.data)`) and handle different `event.event` types (like 'create', 'update', 'delete') according to their application logic.
  Fix: After receiving an event, use `import json` and then `data_payload = json.loads(event.data)` to access the structured data. Implement conditional logic based on `event.event` for proper handling.
• gotcha While the underlying `sseclient-py` library includes basic reconnect logic, robust, long-running applications may require custom retry policies or advanced error handling to maintain stable connections, especially under unreliable network conditions or server-side disconnections.
  Fix: Implement custom retry logic around the `client.events()` loop using libraries like `tenacity` or a `while True` loop with backoff and error logging to ensure continuous operation.
• gotcha Incorrect `base_url` or `api_key` will lead to authentication failures or connection errors, which might manifest as the `events()` generator simply stopping without yielding events, or raising a low-level `requests.exceptions.HTTPError` or `ConnectionError`.
  Fix: Double-check `OPENCTI_BASE_URL` and `OPENCTI_API_KEY` values. Ensure the API key has the necessary permissions. Add explicit `try...except requests.exceptions.HTTPError as e:` blocks to catch authentication-related status codes (e.g., 401, 403).

Install

• pip install filigran-sseclient Install latest version

Imports

• SSEClient

  from filigran_sseclient import SSEClient

Quickstart

This quickstart demonstrates how to connect to an OpenCTI Server-Sent Events stream using `filigran-sseclient`. It initializes an `SSEClient` with a base URL and API key (retrieved from environment variables for security) and then iterates through incoming events. Remember to replace placeholder values with your actual OpenCTI instance details and ensure `OPENCTI_API_KEY` is securely configured.

import os
from filigran_sseclient import SSEClient

# Replace with your OpenCTI instance URL and API key/token
OPENCTI_BASE_URL = os.environ.get('OPENCTI_BASE_URL', 'https://demo.opencti.io')
OPENCTI_API_KEY = os.environ.get('OPENCTI_API_KEY', '') # Ensure this is set securely

if not OPENCTI_API_KEY:
    print("WARNING: OPENCTI_API_KEY environment variable not set. Connection may fail or use a public demo instance.")

print(f"Attempting to connect to OpenCTI SSE at {OPENCTI_BASE_URL}...")

try:
    client = SSEClient(
        base_url=OPENCTI_BASE_URL,
        api_key=OPENCTI_API_KEY,
        verify=False # Set to True in production with proper SSL certificates
    )

    print("Connected. Waiting for events...")
    for event in client.events():
        print(f"Received Event - ID: {event.id}, Type: {event.event}, Data: {event.data}")
        # Process event.event (e.g., 'create', 'update', 'delete')
        # Process event.data (usually JSON string of the OpenCTI object)
        # For demonstration, break after a few events or on a specific condition
        if event.id and int(event.id) > 5: # Example: process first 5 events
            break

except Exception as e:
    print(f"An error occurred during SSE connection or event processing: {e}")
finally:
    print("SSE client connection terminated.")