{"id":7225,"library":"filigran-sseclient","title":"Filigran SSEClient","description":"Filigran SSEClient provides a Python API client specifically designed for consuming Server-Sent Events (SSE) from OpenCTI platforms. It simplifies connection management, event parsing, and authentication for real-time data streaming from OpenCTI. The current version is 1.0.2, and releases are typically made to address bug fixes or minor enhancements.","status":"active","version":"1.0.2","language":"en","source_language":"en","source_url":"https://github.com/FiligranHQ/filigran-sseclient","tags":["sse","events","opencti","client","streaming","real-time"],"install":[{"cmd":"pip install filigran-sseclient","lang":"bash","label":"Install latest version"}],"dependencies":[{"reason":"Core library for handling Server-Sent Events.","package":"sseclient-py","optional":false},{"reason":"HTTP client for making underlying network requests.","package":"requests","optional":false},{"reason":"Utilities for requests, potentially for advanced connection features.","package":"requests-toolbelt","optional":false}],"imports":[{"symbol":"SSEClient","correct":"from filigran_sseclient import SSEClient"}],"quickstart":{"code":"import os\nfrom filigran_sseclient import SSEClient\n\n# Replace with your OpenCTI instance URL and API key/token\nOPENCTI_BASE_URL = os.environ.get('OPENCTI_BASE_URL', 'https://demo.opencti.io')\nOPENCTI_API_KEY = os.environ.get('OPENCTI_API_KEY', '') # Ensure this is set securely\n\nif not OPENCTI_API_KEY:\n    print(\"WARNING: OPENCTI_API_KEY environment variable not set. Connection may fail or use a public demo instance.\")\n\nprint(f\"Attempting to connect to OpenCTI SSE at {OPENCTI_BASE_URL}...\")\n\ntry:\n    client = SSEClient(\n        base_url=OPENCTI_BASE_URL,\n        api_key=OPENCTI_API_KEY,\n        verify=False # Set to True in production with proper SSL certificates\n    )\n\n    print(\"Connected. Waiting for events...\")\n    for event in client.events():\n        print(f\"Received Event - ID: {event.id}, Type: {event.event}, Data: {event.data}\")\n        # Process event.event (e.g., 'create', 'update', 'delete')\n        # Process event.data (usually JSON string of the OpenCTI object)\n        # For demonstration, break after a few events or on a specific condition\n        if event.id and int(event.id) > 5: # Example: process first 5 events\n            break\n\nexcept Exception as e:\n    print(f\"An error occurred during SSE connection or event processing: {e}\")\nfinally:\n    print(\"SSE client connection terminated.\")","lang":"python","description":"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."},"warnings":[{"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.","message":"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.","severity":"gotcha","affected_versions":"All versions"},{"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.","message":"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.","severity":"gotcha","affected_versions":"All versions"},{"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).","message":"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`.","severity":"gotcha","affected_versions":"All versions"}],"env_vars":null,"last_verified":"2026-04-16T00:00:00.000Z","next_check":"2026-07-15T00:00:00.000Z","problems":[{"fix":"Run `pip install filigran-sseclient` in your active Python environment.","cause":"The `filigran-sseclient` package is not installed or the Python environment is incorrect.","error":"ModuleNotFoundError: No module named 'filigran_sseclient'"},{"fix":"Change `token=your_token` to `api_key=your_api_key` when initializing `SSEClient`.","cause":"The constructor for `SSEClient` expects `api_key` for authentication, not `token`.","error":"TypeError: SSEClient.__init__() got an unexpected keyword argument 'token'"},{"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).","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.","error":"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'))"}]}