Lumos Connector Quick Reference Guide

This guide serves as a quick reference for developers building custom Lumos connectors. It explains each file in a logical order to help you understand the connector structure and implement your own connector efficiently.

Implementation Workflow

When implementing a connector, follow this recommended order:

Define settings in settings.py - what configuration does your connector need?
Declare credentials in auth.py - which credentials does your app require?
Define resources and entitlements in enums.py - map your app's concepts to Lumos
Set up constants in constants.py - define API URLs and credential IDs
Implement the client in client.py - create methods for API communication
Implement validate_credentials in capabilities_read.py - this is the minimum required capability
Implement read capabilities in capabilities_read.py - typically easier to implement
Implement write capabilities in capabilities_write.py - as needed
Configure the integration in integration.py - register your implemented capabilities

Minimum Viable Connector

A minimal connector needs:

settings.py - Defining what configuration users need to provide
auth.py - Declaring at least one credential
constants.py - Defining the credential ID enum and base URL
enums.py - Defining resource and entitlement types (can be minimal)
client.py - With at least enough functionality to validate credentials
capabilities_read.py - With at least validate_credentials implemented
integration.py - Registering at least the validate_credentials capability

Core Files Overview

Here's a conceptual map of how the files relate to each other:

integration.py  <-- The central configuration file
    |
    ├── auth.py        <-- Credentials the connector accepts
    ├── settings.py    <-- User configuration parameters
    ├── enums.py       <-- Resource and entitlement types
    ├── constants.py   <-- API URLs and credential IDs
    ├── client.py      <-- HTTP client(s) for API communication
    │        |
    │        ├── capabilities_read.py  <-- Read operations implementation
    │        └── capabilities_write.py <-- Write operations implementation
    └── pagination.py  <-- Pagination utilities

1. `settings.py`

Purpose: Defines the configuration settings that users will provide when setting up the connector.

Key aspects:

Uses Pydantic models for type safety and validation
Each setting can have descriptions, defaults, and validation rules
Settings are passed to every capability function
Authentication secrets belong in auth.py, not here

Example implementation:

from pydantic import BaseModel, Field

class MyAppSettings(BaseModel):
    """Settings for MyApp connector."""
    api_url: str = Field(
        description="The base URL of the MyApp API"
    )
    use_https: bool = Field(
        default=True,
        description="Whether to use HTTPS for API calls"
    )

2. `constants.py`

Purpose: Stores constant values used throughout the connector, like API paths.

Key aspects:

API endpoints and base URLs
The credential ID enum - one entry per credential declared in auth.py
Default values and URL templates

Credential IDs are part of the connector's public contract. Renaming one invalidates existing connections, so choose them carefully and keep them in one place.

Example implementation:

from enum import Enum

# API endpoints
BASE_URL = "https://api.myapp.com"
API_BASE_PATH = "/api"
USER_ENDPOINT = f"{API_BASE_PATH}/users"

class MyAppCredentialId(str, Enum):
    OAUTH = "myapp_oauth"
    SCIM = "myapp_scim"

3. `auth.py`

Purpose: Declares every credential the connector accepts, with the setup instructions
shown to the user.

Key aspects:

One CredentialConfig per credential, or OAuthConfig for OAuth flows
Markdown description per credential
optional=True for credentials the connector can run without
validation= to attach a cheap validation check on the credential

Example implementation:

from connector.oai.capability import StandardCapabilityName
from connector.oai.modules.oauth_module_types import OAuthSettings
from connector_sdk_types import AuthModel, CredentialConfig, OAuthConfig

from myapp_connector.constants import MyAppCredentialId

OAUTH_DESCRIPTION = """To set up OAuth with MyApp:

1. Open Settings -> Developer -> OAuth apps
2. Add `https://app.lumosidentity.com/integrations/ics_oauth2_callback` as a redirect URI
3. Copy the Client ID and Client Secret into the fields below
"""

SCIM_DESCRIPTION = """SCIM token, used for provisioning. Generate one under
Settings -> Provisioning -> SCIM.
"""

MyAppCredentialsConfig = [
    OAuthConfig(
        id=MyAppCredentialId.OAUTH,
        type=AuthModel.OAUTH,
        description=OAUTH_DESCRIPTION,
        oauth_settings=OAuthSettings(
            authorization_url="https://myapp.com/oauth/authorize",
            token_url="https://myapp.com/oauth/token",
            scopes={
                StandardCapabilityName.VALIDATE_CREDENTIALS: "users:read",
                StandardCapabilityName.LIST_ACCOUNTS: "users:read",
            },
        ),
    ),
    CredentialConfig(
        id=MyAppCredentialId.SCIM,
        name="SCIM Token",
        type=AuthModel.TOKEN,
        description=SCIM_DESCRIPTION,
        optional=True,
    ),
]

See Authorization and Authentication for the full pattern, including per-credential validators and custom input models.

4. `enums.py`

Purpose: Defines the types of resources and entitlements in your application.

Key aspects:

Resource types (containers like projects, teams, etc.)
Entitlement types (permissions within resources)
Maps application-specific concepts to Lumos concepts
EntitlementType requires min; add max where the app caps assignments

Example implementation:

from enum import Enum
from connector.generated import EntitlementType, ResourceType

class MyAppResourceTypes(str, Enum):
    PROJECT = "project"
    TEAM = "team"

class MyAppEntitlementTypes(str, Enum):
    PROJECT_ADMIN = "project_admin"
    TEAM_MEMBER = "team_member"

resource_types: list[ResourceType] = [
    ResourceType(
        type_id=MyAppResourceTypes.PROJECT,
        type_label="Project",
    ),
    # ...
]

entitlement_types: list[EntitlementType] = [
    EntitlementType(
        type_id=MyAppEntitlementTypes.PROJECT_ADMIN,
        type_label="Project Administrator",
        resource_type_id=MyAppResourceTypes.PROJECT,
        min=0,
    ),
    # ...
]

5. `pagination.py`

Purpose: Handles pagination for large data sets.

Key aspects:

Stores pagination state between requests
Encodes/decodes pagination tokens
Maintains default page sizes
paginations_from_args is the helper you call from a capability

Example implementation:

import typing as t

from connector.client import (
    NextPageTokenInterface,
    PaginationBase,
    create_next_page_token,
    get_page,
)
from connector.oai.capability import Request

DEFAULT_PAGE_SIZE = 25

class Pagination(PaginationBase):
    """Pagination parameters for API methods."""
    offset: int

    @classmethod
    def default(cls, endpoint: str) -> "Pagination":
        return cls(
            endpoint=endpoint,
            offset=0,
        )

if t.TYPE_CHECKING:
    class NextPageToken(NextPageTokenInterface[Pagination]):  # pragma: no cover
        @classmethod
        def from_paginations(cls, paginations: list[Pagination]) -> "NextPageToken":
            return cls(token=None)

        def paginations(self) -> list[Pagination]:
            return []
else:
    NextPageToken = create_next_page_token(Pagination, "NextPageToken")


def paginations_from_args(
    args: Request, default_endpoints: list[str]
) -> tuple[list[Pagination], Pagination, int]:
    paginations = NextPageToken(get_page(args).token).paginations()
    if not paginations:
        paginations = [Pagination.default(endpoint) for endpoint in default_endpoints]

    current_pagination = paginations.pop()
    return paginations, current_pagination, get_page(args).size or DEFAULT_PAGE_SIZE

The scaffold generates this file for you; you normally only change the Pagination fields (offset, cursor, page, …) to match your API's pagination style.

6. `client.py`

Purpose: Handles communication with your application's API.

Key aspects:

Authentication with your API
Methods for each API endpoint
Always call response.raise_for_status() so the SDK's exception handlers can turn
API failures into typed connector errors
Error handling for API calls
Transformation of API responses

Example implementation:

import typing as t

from connector.generated import OAuthCredential
from connector.oai.base_clients import BaseIntegrationClient
from connector.oai.capability import Request, get_credential, get_settings
from connector.utils.httpx_auth import BearerAuth

from myapp_connector.constants import API_BASE_PATH, MyAppCredentialId
from myapp_connector.settings import MyAppSettings

class MyAppClient(BaseIntegrationClient):
    @classmethod
    def prepare_client_args(cls, args: Request) -> dict[str, t.Any]:
        """Configure the HTTP client with auth and base URL."""
        settings = get_settings(args, MyAppSettings)
        oauth = get_credential(args, MyAppCredentialId.OAUTH, OAuthCredential)
        return {
            "auth": BearerAuth(token=oauth.access_token),
            "base_url": settings.api_url,
        }

    async def get_users(
        self, limit: int | None = None, offset: int | None = None
    ) -> dict[str, t.Any]:
        """Fetch users from the API with pagination."""
        params: dict[str, t.Any] = {}
        if limit is not None:
            params["limit"] = limit
        if offset is not None:
            params["offset"] = offset

        response = await self._http_client.get(f"{API_BASE_PATH}/users", params=params)
        response.raise_for_status()
        return t.cast(dict[str, t.Any], response.json())

    # Additional methods for other API endpoints

7. `capabilities_read.py`

Purpose: Implements read operations (list_accounts, list_resources, etc.).

Key aspects:

Translates between your API and Lumos data formats
Handles pagination
Lets registered exception handlers deal with API errors rather than catching broadly

Example implementation:

from connector.generated import (
    AccountStatus,
    FoundAccountData,
    ListAccountsRequest,
    ListAccountsResponse,
)

from myapp_connector.client import MyAppClient
from myapp_connector.pagination import NextPageToken, Pagination, paginations_from_args

USERS_ENDPOINT = "/api/users"

async def list_accounts(args: ListAccountsRequest) -> ListAccountsResponse:
    """List user accounts from the application."""
    paginations, current_pagination, page_size = paginations_from_args(
        args, default_endpoints=[USERS_ENDPOINT]
    )

    async with MyAppClient(args) as client:
        response = await client.get_users(
            limit=page_size,
            offset=current_pagination.offset,
        )

        accounts = [
            FoundAccountData(
                integration_specific_id=user["id"],
                email=user.get("email"),
                username=user.get("email"),
                given_name=user.get("name"),
                user_status=(
                    AccountStatus.ACTIVE
                    if user.get("active", True)
                    else AccountStatus.INACTIVE
                ),
            )
            for user in response.get("items", [])
        ]

        if response.get("has_more", False):
            paginations.append(
                Pagination(
                    endpoint=current_pagination.endpoint,
                    offset=current_pagination.offset + len(accounts),
                )
            )

    return ListAccountsResponse(
        response=accounts,
        page=NextPageToken.from_paginations(paginations).to_page(page_size)
        if paginations
        else None,
    )

Note the field names on FoundAccountData: the account identifier is
integration_specific_id (not account_id) and the status field is user_status,
which takes an AccountStatus enum member.

8. `capabilities_write.py`

Purpose: Implements write operations (create_account, assign_entitlement, etc.).

Key aspects:

Translates Lumos requests to your API format
Handles creation, updates, and deletion of entities
create_account takes a CustomRequest[CreateAccount], because the fields needed to
create an account differ per app - they are declared in dto/user.py

Example implementation:

from connector.generated import (
    AccountStatus,
    CreateAccountResponse,
    CreatedAccount,
)
from connector.oai.capability import CustomRequest

from myapp_connector.client import MyAppClient
from myapp_connector.dto.user import CreateAccount

async def create_account(args: CustomRequest[CreateAccount]) -> CreateAccountResponse:
    """Create a new user account in the application."""
    request = args.request

    async with MyAppClient(args) as client:
        new_user = await client.create_user(
            {
                "email": request.email,
                "name": request.display_name,
                "status": "active",
            }
        )

    return CreateAccountResponse(
        response=CreatedAccount(
            id=new_user["id"],
            status=AccountStatus.ACTIVE,
            created=True,
        )
    )

9. `integration.py`

Purpose: The main configuration file that registers capabilities and sets up the connector.

Key aspects:

Defines connector metadata (name, description, logo)
Passes the credential list from auth.py as credentials
Declares valid credential combinations in credentials_settings
Registers capabilities and configures error handling

Example implementation:

import httpx
from connector.generated import AppCategory, StandardCapabilityName
from connector.oai.errors import HTTPHandler
from connector.oai.integration import DescriptionData, Integration
from connector_sdk_types import CredentialsSettings

from myapp_connector import capabilities_read, capabilities_write
from myapp_connector.__about__ import __version__
from myapp_connector.auth import MyAppCredentialsConfig
from myapp_connector.constants import MyAppCredentialId
from myapp_connector.enums import entitlement_types, resource_types
from myapp_connector.settings import MyAppSettings

integration = Integration(
    app_id="myapp",
    version=__version__,
    credentials=MyAppCredentialsConfig,
    credentials_settings=CredentialsSettings(
        allowed_credentials=[
            (MyAppCredentialId.OAUTH,),
            (MyAppCredentialId.OAUTH, MyAppCredentialId.SCIM),
        ],
    ),
    exception_handlers=[
        (httpx.HTTPStatusError, HTTPHandler, None),
    ],
    description_data=DescriptionData(
        logo_url="https://logo.clearbit.com/myapp.com",
        user_friendly_name="MyApp",
        description="MyApp is a cloud-based platform for...",
        categories=[AppCategory.DEVELOPERS, AppCategory.COLLABORATION],
    ),
    settings_model=MyAppSettings,
    resource_types=resource_types,
    entitlement_types=entitlement_types,
)

# Register capabilities
integration.register_capabilities(
    {
        # Required capability
        StandardCapabilityName.VALIDATE_CREDENTIALS: capabilities_read.validate_credentials,

        # Read capabilities
        StandardCapabilityName.LIST_ACCOUNTS: capabilities_read.list_accounts,
        StandardCapabilityName.LIST_RESOURCES: capabilities_read.list_resources,
        StandardCapabilityName.LIST_ENTITLEMENTS: capabilities_read.list_entitlements,

        # Write capabilities
        StandardCapabilityName.CREATE_ACCOUNT: capabilities_write.create_account,
        StandardCapabilityName.ASSIGN_ENTITLEMENT: capabilities_write.assign_entitlement,
    }
)

validate_credential_config is registered for you automatically, so you do not list it
here - but you do need a test case file for it (see Testing below).

10. `main.py`

Purpose: Entry point for the CLI tool.

Key aspects:

Very simple - just runs the integration

Example implementation:

from connector.cli import run_integration

from myapp_connector.integration import integration

def main():
    run_integration(integration)

if __name__ == "__main__":
    main()

10. `about.py`

Purpose: Contains version information.

Example implementation:

__version__ = "0.1.0"

Testing Your Connector

Run these commands to test your connector:

# Typecheck your code
mypy .

# Run unit tests
pytest

# Test validate_credentials
myapp-connector validate_credentials --json '{"credentials":[{"id":"myapp_oauth","token":{"token":"..."}}],"request":{},"settings":{"api_url":"https://api.myapp.com"}}'

Requests pass credentials as a credentials array, where each entry is tagged with the
ID it was declared under in auth.py and nests its payload under the key matching its
authentication model (oauth, token, basic, …).

About the generated test cases. The scaffold creates a case file per capability, but
they are placeholders wired to a fictional /example endpoint. Once you implement real
capabilities, those placeholders fail until you replace their mocked URLs and expected
responses with your own. Two rules to know:

Case files must be named test_{capability_name}_cases.py, or the runner will not find them.
Every registered capability needs a case file, including validate_credential_config.

Authentication Models

Declare each credential in auth.py with one of these models:

Authentication model	Payload model	Typical use
`AuthModel.OAUTH`	`OAuthCredential`	OAuth 2.0 authorization code
`AuthModel.OAUTH_CLIENT_CREDENTIALS`	`OAuthClientCredential`	OAuth 2.0 machine-to-machine
`AuthModel.TOKEN`	`TokenCredential`	API keys and bearer tokens
`AuthModel.BASIC`	`BasicCredential`	Username and password
`AuthModel.JWT`	`JWTCredential`	JWT-based auth
`AuthModel.SERVICE_ACCOUNT`	`ServiceAccountCredential`	Service account key files
`AuthModel.KEY_PAIR`	`KeyPairCredential`	Public/private key pairs

Key Capabilities

Here are the most important capabilities to implement:

validate_credentials (Required) - Verifies connection and credentials
list_accounts - Lists user accounts
list_resources - Lists available resources (projects, teams, etc.)
list_entitlements - Lists available entitlements (roles, permissions)
find_entitlement_associations - Shows which users have which entitlements

The write capabilities are optional but useful:

create_account - Creates a new user account
assign_entitlement - Grants access/permissions to a user
unassign_entitlement - Removes access/permissions from a user

Troubleshooting Tips

A broken connection reports as valid: you are missing response.raise_for_status() in your client methods
Cannot find implementation or library stub for module named "connector...": install the SDK into the same environment you run mypy from; an editable install of the SDK cannot be followed by mypy
Authentication issues: check the credential ID in your request matches the ID declared in auth.py, and that the payload is nested under the right key
get_credential raises for a credential you marked optional: pass strict=False and handle None
Missing data: verify field mappings between your API and Lumos models - integration_specific_id, user_status, label
Pagination errors: ensure your Pagination fields match your API's paging style
Type errors: run mypy . to catch type mismatches

Next Steps

After implementing the basic connector:

Add more capabilities - Implement additional read/write capabilities
Improve error handling - Add specific exception handlers
Add custom attributes - For application-specific user properties
Add a second credential - If your app needs SCIM or a separate admin API

For more detailed guidance, refer to the full documentation or the complete tutorial.

Implementation Workflow

Minimum Viable Connector

Core Files Overview

1. settings.py

2. constants.py

3. auth.py

4. enums.py

5. pagination.py

6. client.py

7. capabilities_read.py

8. capabilities_write.py

9. integration.py

10. main.py

10. __about__.py

Testing Your Connector

Authentication Models

Key Capabilities

Troubleshooting Tips

Next Steps

1. `settings.py`

2. `constants.py`

3. `auth.py`

4. `enums.py`

5. `pagination.py`

6. `client.py`

7. `capabilities_read.py`

8. `capabilities_write.py`

9. `integration.py`

10. `main.py`

10. `about.py`