OpenCTI Connector Manager Explained: How to Register Custom Connectors

The Connector Manager is a background service that periodically cleans up stale connector works and reports health status, while custom connectors register via GraphQL mutations, ingestion helpers, or the Python SDK to create RabbitMQ queues and Connector entities.

The OpenCTI connector manager orchestrates background task cleanup and connector lifecycle management within the threat intelligence platform. Understanding how this component works and how to properly register custom connectors is essential for developers extending the OpenCTI ecosystem. In the OpenCTI-Platform/opencti repository, the manager runs as a scheduled service inside the API process, coordinating with Redis and RabbitMQ to maintain system health.

What Is the Connector Manager?

The Connector Manager is a background service that runs inside the OpenCTI API process. It is instantiated once at server startup and exported from src/manager/connectorManager.js (line 165) as a singleton with public methods for start, status, and shutdown.

Core Responsibilities

The manager operates under a distributed locking mechanism to ensure only one API instance executes it at a time. It acquires the CONNECTOR_MANAGER_KEY lock via lockResources (lines 11-15) before performing four critical maintenance tasks:

1. Periodic Execution The manager runs every connector_manager:interval (default 60 seconds) using setIntervalAsync created at lines 43-46 in src/manager/connectorManager.js.

2. Closing Old Works The closeOldWorks function (lines 21-78) reads connector status from Redis, then queries for Work documents matching connector_id + status (wait/progress) + timestamp < current. It marks these as complete when a newer work exists for the same connector.

3. Cleaning Completed Works The deleteCompletedWorks function (lines 80-107) removes works that are already complete and older than connector_manager:works_day_range (default 7 days), deleting both the Elasticsearch documents and their associated Redis entries.

4. Status Reporting Health is exposed via the GraphQL ConnectorManager type. In src/resolvers/connector.js (lines 124-126), the active field checks whether the manager has executed within the last 5 minutes.

// src/manager/connectorManager.js
const initConnectorManager = () => ({
  start: async () => {
    scheduler = setIntervalAsync(connectorHandler, SCHEDULE_TIME);
  },
  status: async () => ({
    id: 'CONNECTOR_MANAGER',
    enable: booleanConf('connector_manager:enabled', false),
    running,
  }),
  shutdown: async () => {
    if (scheduler) return clearIntervalAsync(scheduler);
    return true;
  },
});

How to Register Custom Connectors in OpenCTI

A custom connector is represented as an entity of type Connector in the platform. Registration creates the necessary RabbitMQ queues and establishes the connector's identity in Elasticsearch. There are three primary registration patterns:

Method 1: GraphQL Mutation (Manual Registration)

Use the registerConnector mutation when you have a running connector process and need the platform to recognize it. The resolver in src/resolvers/connector.js (line 41) forwards to registerConnector in src/domain/connector.ts (lines 312-380).

mutation RegisterMyConnector {
  registerConnector(
    input: {
      id: "a2de809c-fbb9-491d-90c0-96c7d1766000"
      name: "My Awesome Connector"
      type: EXTERNAL_IMPORT
      scope: ["application/pdf"]
      auto: true
      auto_update: false
      only_contextual: false
      playbook_compatible: false
      listen_callback_uri: null
    }
  ) {
    id
    name
    connector_type
    connector_scope
    active
  }
}

Under the hood, this calls registerConnectorQueues (line 22) to set up RabbitMQ infrastructure, then creates or updates the Connector entity via createEntity, publishes a User Action event, and invalidates the cache.

Method 2: Automatic Registration via Ingestion Helpers

When creating a Feed (RSS, TAXII, etc.) through the UI, the platform automatically calls registerConnectorForIngestion in src/domain/connector.ts (lines 997-1013).

export const registerConnectorForIngestion = async (context, input) => {
  await registerConnector(context, SYSTEM_USER, {
    id: connectorIdFromIngestId(input.id),
    name: `[FEED - ${input.type}] ${input.name}`,
    type: ConnectorType.ExternalImport,
    auto: true,
    auto_update: false,
    scope: ['application/stix+json;version=2.1'],
    only_contextual: false,
    playbook_compatible: false,
  }, {
    built_in: true,
    active: input.is_running,
    connector_user_id: input.connector_user_id,
  });
};

This generates a deterministic internal ID via connectorIdFromIngestId and sets built_in: true, preventing deletion from the UI while linking the connector to the ingestion source.

Method 3: Python SDK Registration

The pycti library provides OpenCTIConnectorHelper.register_connector() to perform the GraphQL mutation programmatically.

from pycti import OpenCTIConnectorHelper

config = {
    "url": "https://my-opencti.example.com",
    "token": "MY_TOKEN",
    "connector_id": "a2de809c-fbb9-491d-90c0-96c7d1766000",
    "connector_name": "My Awesome Connector",
    "connector_type": "EXTERNAL_IMPORT",
    "connector_scope": "application/pdf",
}

helper = OpenCTIConnectorHelper(config)
helper.register_connector()

This wrapper handles HTTP requests, authentication, and error handling while executing the same GraphQL mutation as the manual method.

Connector Manager Lifecycle Operations

You can programmatically control the manager using the exported default from src/manager/connectorManager.js:

import connectorManager from '../manager/connectorManager';

// Start (normally automatic at boot)
await connectorManager.start();

// Check operational status
const status = await connectorManager.status();
console.log('Manager running:', status.running);

// Graceful shutdown during server stop
await connectorManager.shutdown();

Custom Connector Implementation Example

Here is a complete skeleton for building a Python-based custom connector that registers itself and processes messages:

import os
import yaml
from pycti import OpenCTIConnectorHelper

class MyConnector:
    def __init__(self):
        config_path = os.path.join(os.path.dirname(__file__), "config.yml")
        with open(config_path) as f:
            config = yaml.safe_load(f)
        self.helper = OpenCTIConnectorHelper(config)

    def run(self):
        # Register or update connector entity

        self.helper.register_connector()
        
        # Listen for work messages from the platform

        self.helper.listen(self._process_message)

    def _process_message(self, data):
        self.helper.log_info(f"Received work: {data}")
        # Process the data...

        # Use self.helper.send_stix2_bundle() to return results

if __name__ == "__main__":
    connector = MyConnector()
    connector.run()

This implementation leverages the OpenCTIConnectorHelper class from the client-python repository to handle registration, message queuing, and STIX bundle transmission automatically.

Summary

• The Connector Manager in src/manager/connectorManager.js is a scheduled background service that closes old works, deletes completed works older than 7 days, and reports health status via GraphQL.
• It uses distributed locking (CONNECTOR_MANAGER_KEY) to ensure single-instance execution across API replicas.
• Register custom connectors using the registerConnector GraphQL mutation, the registerConnectorForIngestion helper for feeds, or the Python SDK's register_connector() method.
• Registration creates RabbitMQ queues via registerConnectorQueues and persists a Connector entity in Elasticsearch.
• The manager tracks connector activity through Redis and ensures work items are properly cleaned up to prevent storage bloat.

Frequently Asked Questions

What does the OpenCTI connector manager do?

The connector manager runs every 60 seconds (configurable via connector_manager:interval) to maintain system hygiene. It closes stale "wait" or "progress" works when newer works exist, permanently deletes completed works older than the configured retention period (default 7 days), and exposes an active status field through the GraphQL API to indicate operational health.

How do I register a custom connector manually?

Execute the registerConnector GraphQL mutation against the OpenCTI API, providing a unique UUID, connector name, type (such as EXTERNAL_IMPORT or INTERNAL_ENRICHMENT), and scope array defining the MIME types your connector handles. The mutation resolver in src/resolvers/connector.js processes this request by calling the domain logic in src/domain/connector.ts, which creates the RabbitMQ queues and Elasticsearch entity.

What is the difference between manual and ingestion-based connector registration?

Manual registration via GraphQL or the Python SDK creates standalone connectors that appear in the UI and can be managed independently. Ingestion-based registration via registerConnectorForIngestion creates hidden, built-in connectors linked to specific feeds (RSS, TAXII, CSV) that cannot be deleted separately from their feed configuration and automatically derive their scope from the ingestion type.

How can I verify that the connector manager is running?

Query the GraphQL API for the connectorManager field, which checks whether the manager executed within the last 5 minutes. Alternatively, inspect the API logs for successful acquisition of the CONNECTOR_MANAGER_KEY lock and execution of closeOldWorks and deleteCompletedWorks functions from src/manager/connectorManager.js.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}