Guides

🔌 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:

# 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:

# 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, we modified 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)

We created 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:

# 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:

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:

  1. Maintain proper type hints and documentation
  2. Implement comprehensive error handling
  3. Use appropriate status mapping for your system
  4. Consider implementing rate limiting for production use
  5. Add logging for better observability
  6. Include comprehensive test coverage
  7. 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.

Updated 10 days ago

Did this page help you?