> ## Documentation Index
> Fetch the complete documentation index at: https://docs.lumos.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Connector SDK

<Icon src="https://img.icons8.com/external-flaticons-lineal-color-flat-icons/64/external-integration-web-development-flaticons-lineal-color-flat-icons.png" /> 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:

```c bash theme={null}
# 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:

```c bash theme={null}
# 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](/reference/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:

```bash theme={null}
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<br />`credentials` field, and `tests/common_mock_data.py` defines them tagged with their<br />credential IDs:

```python theme={null}
# 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`)

```python theme={null}
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:

```python theme={null}
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:

```python theme={null}
# 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:

```python theme={null}
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 bash theme={null}
# 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 bash theme={null}
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:

```bash theme={null}
# 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](https://pypi.org/project/connector-py/#history) for version history.

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