Connector SDK

Sample integration built using Connector SDK

Introduction

This guide demonstrates how to build a simple read-only Okta connector using the Lumos Connector SDK. The connector implements user synchronization functionality, mapping Okta user profiles to Lumos account structures. Note: This sample Okta connector has been created for demonstration purposes, Lumos provides a full-featured native Okta integration.

Prerequisites

Before starting development, ensure you have:

Python 3.10 or higher installed
Access to an Okta developer account
An Okta API token
Basic familiarity with Python and async programming

Setting Up the Development Environment

First, create and activate a Python virtual environment:

bash

# Create a new directory for your project
mkdir okta
cd okta

# Create a Python virtual environment
python -m venv .venv

# Activate the virtual environment# On Windows:
.venv\Scripts\activate
# On Unix/MacOS:
source .venv/bin/activate

# Install the Lumos Connector SDK with development dependencies
pip install "connector-py[dev]"

Generating the Facade Connector

Use the Lumos Connector SDK to generate a basic connector structure:

bash

# Generate the connector scaffold
connector scaffold okta okta

# Install the connector's dependencies and the connector module
# It is important to use -e (editable) here so code changes will be reflected when running commands
cd okta
pip install -e ".[all]"

# Verify the installation
mypy .
pytest

The scaffold generates a multi-auth connector: it declares the credentials it accepts in an auth.py file and passes them to the Integration as credentials. Out of the box you get a single OAuth credential with a validate_credential_config validator, which you then adapt to whatever the target app actually needs — a token, basic auth, or several credentials side by side. See Authorization and Authentication for the full pattern. If you are working on a connector that only ever takes one credential and you want the older single-credential shape (auth=OAuthCredential, read with get_oauth(args)), pass --single-auth to the scaffold tool:

connector scaffold okta okta --single-auth

New connectors (Q2 2026 onward) should prefer the default. The single-credential shape cannot express more than one credential or per-credential validation.

Understanding the Generated Connector Structure

The scaffold command creates several key files:

Core Files

__about__.py: Contains version information
main.py: Entry point for the connector
auth.py: Declares the credentials the connector accepts, and their validators
settings.py: Defines connector configuration model
integration.py: Defines connector capabilities and metadata
client.py: Implements API client functionality
constants.py: Stores shared constants, including the credential ID enum and OAuth URLs
enums.py: Defines resource and entitlement types

auth.py is only generated for multi-auth connectors. With --single-auth, the credential is declared inline in integration.py instead.

Capability Implementation Files

capabilities_read.py: Implements read operations
capabilities_write.py: Implements write operations (not used in this example)

Testing Files

test_all_capabilities.py: Main test runner
test_read_capabilities/test_validate_credential_config_cases.py: Cases for the per-credential validation capability, which the SDK registers automatically for multi-auth connectors
Various test case files for different capabilities

Test cases for a multi-auth connector attach credentials as a list on the request’s
credentials field, and tests/common_mock_data.py defines them tagged with their
credential IDs:

# Example: tests/common_mock_data.py
from connector.generated import AuthCredential, OAuthCredential

from okta.constants import OktaCredentialId

VALID_AUTH = [
    AuthCredential(
        id=OktaCredentialId.OAUTH,
        oauth=OAuthCredential(access_token="valid-access-token"),
    ),
]

Implementing the Okta Connector

To transform the facade connector into a functional Okta connector, modify several files:

1. Settings Model (`settings.py`)

from pydantic import BaseModel

class OktaSettings(BaseModel):
    domain: str# Okta domain e.g. "company.okta.com"

2. Type Definitions (`okta_types.py`)

Create this new file to handle Okta-specific types and status mappings:

from enum import Enum
from typing import TypedDict

class OktaUserStatus(str, Enum):
    ACTIVE = "ACTIVE"
    SUSPENDED = "SUSPENDED"
# ... other status values

def map_okta_to_lumos_status(okta_status: str) -> str:
# ... status mapping implementation

3. Integration Configuration (`integration.py`)

Updated to use token authentication and register appropriate capabilities:

# okta/auth.py
from connector_sdk_types import AuthModel, CredentialConfig
from okta.constants import OktaCredentialId

OktaCredentialsConfig = [
    CredentialConfig(
        id=OktaCredentialId.API_TOKEN,
        name="API Token",
        type=AuthModel.TOKEN,
        description="Create an API token in the Okta admin console under Security -> API -> Tokens.",
    ),
]

# okta/integration.py
integration = Integration(
    app_id="okta",
    credentials=OktaCredentialsConfig,
    credentials_settings=CredentialsSettings(
        allowed_credentials=[(OktaCredentialId.API_TOKEN,)],
    ),
    settings_model=OktaSettings,
    # ... other configuration
)

integration.register_capabilities({
    StandardCapabilityName.VALIDATE_CREDENTIALS: capabilities_read.validate_credentials,
    StandardCapabilityName.LIST_ACCOUNTS: capabilities_read.list_accounts,
})

4. Client Implementation (`client.py`)

Implemented the Okta API client with proper authentication and user mapping:

class OktaConnectorClient(BaseIntegrationClient):
    @classmethod
    def prepare_client_args(cls, args: Request) -> dict[str, t.Any]:
        token = get_credential(args, OktaCredentialId.API_TOKEN, TokenCredential)
        settings = get_settings(args, OktaSettings)
        return {
            "auth": SSWSAuth(token=token.token),
            "base_url": f"https://{settings.domain}",
        }

    async def list_users(self, limit: int | None = None,
                        offset: int | None = None) -> list[FoundAccountData]:
        ...  # implementation of user listing and mapping

Implementation Details

User Status Mapping

The connector maps Okta user statuses to Lumos account statuses:

ACTIVE → ACTIVE
SUSPENDED → SUSPENDED
DEPROVISIONED → DEPROVISIONED
STAGED → PENDING
Others → INACTIVE

Pagination

The implementation supports pagination using Okta’s offset-based pagination mechanism, allowing the connector to handle large user sets efficiently.

Error Handling

The connector implements proper error handling for API failures and includes type validation to ensure data consistency.

Testing the Connector

Execute the test suite to verify the implementation:

bash

# Run type checking
mypy .

# Run unit tests
pytest

# Test specific capabilities
okta validate_credentials --json '{"request":{},"settings":{"domain":"test-domain.okta.com"},"credentials":[{"id":"okta_api_token","token":{"token":"test-token"}}]}'

Expected output:

bash

INFO      HTTP Request: GET https://test-domain.okta.com/api/v1/users?limit=1 "HTTP/1.1 200 OK"
INFO      Result printing to console
{"response":{"valid":true,"unique_tenant_id":"okta_test-domain.okta.com"},"raw_data":null,"page":null}
INFO      Command completed

Deployment

To deploy the connector:

# Build the on-premise connector
connector compile-on-prem --connector-root-module-dir ./okta/okta --app-id okta

The compiled connector (e.g., okta-0.1.0.tar.gz) can then be deployed to the Lumos on-premise agent environment by copying it to the connectors folder (e.g., /lumos-on-premise-agent/connectors) After a minute, the connector will be available for configuration in the Lumos administrative user interface under Integrations where it can be configured. Once connected, it will show confirmation. A few minutes later, after the import is completed, the users imported via this connector will be shown in the Apps page.

Best Practices and Considerations

When implementing your own connector:

Maintain proper type hints and documentation
Implement comprehensive error handling
Use appropriate status mapping for your system
Consider implementing rate limiting for production use
Add logging for better observability
Include comprehensive test coverage
Pin your version of the Connector SDK. This means that in the generated pyproject.toml, you ensure that references to connector-py include a specific version. For example: "connector-py == 4.18.0". Check pypi for version history.

The complete implementation is available for reference, demonstrating these best practices while maintaining clean, maintainable code.

Quickstart

Lumos Connectors

Lumos

Guides

Webhooks

Examples

Tools

Recipes

Connector SDK

Introduction

Prerequisites

Setting Up the Development Environment

Generating the Facade Connector

Understanding the Generated Connector Structure

Core Files

Capability Implementation Files

Testing Files

Implementing the Okta Connector

1. Settings Model (`settings.py`)

2. Type Definitions (`okta_types.py`)

3. Integration Configuration (`integration.py`)

4. Client Implementation (`client.py`)

Implementation Details

User Status Mapping

Error Handling

Testing the Connector

Deployment

Best Practices and Considerations

​Introduction

​Prerequisites

​Setting Up the Development Environment

​Generating the Facade Connector

​Understanding the Generated Connector Structure

​Core Files

​Capability Implementation Files

​Testing Files

​Implementing the Okta Connector

​1. Settings Model (settings.py)

​2. Type Definitions (okta_types.py)

​3. Integration Configuration (integration.py)

​4. Client Implementation (client.py)

​Implementation Details

​User Status Mapping

​Pagination

​Error Handling

​Testing the Connector

​Deployment

​Best Practices and Considerations

Introduction

Prerequisites

Setting Up the Development Environment

Generating the Facade Connector

Understanding the Generated Connector Structure

Core Files

Capability Implementation Files

Testing Files

Implementing the Okta Connector

1. Settings Model (`settings.py`)

2. Type Definitions (`okta_types.py`)

3. Integration Configuration (`integration.py`)

4. Client Implementation (`client.py`)

Implementation Details

User Status Mapping

Pagination

Error Handling

Testing the Connector

Deployment

Best Practices and Considerations