Quick Start Guide

This guide serves as a quick reference for developers building custom Lumos connectors. It explains each file in a logical order so you can 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?
Define resources and entitlements in enums.py - map your app’s concepts to Lumos
Set up constants in constants.py - define API URLs and other constants
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
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
    |
    ├── settings.py  <-- User configuration parameters
    ├── enums.py  	 <-- Resource and entitlement types
    ├── constants.py <-- API URLs and other constants
    ├── client.py  	 <-- HTTP client 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

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
Default values
URL templates

Example implementation:

# API endpoints
API_BASE_PATH = "/api"
USER_ENDPOINT = f"{API_BASE_PATH}/users"
RESOURCES_ENDPOINT = f"{API_BASE_PATH}/resources"

3. `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

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 = [
    ResourceType(
        type_id=MyAppResourceTypes.PROJECT,
        type_label="Project",
    ),
    # ...
]

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

4. `pagination.py`

Purpose: Handles pagination for large data sets. Key aspects:

Stores pagination state between requests
Encodes/decodes pagination tokens
Maintains default page sizes

Example implementation:

from connector.utils.pagination import (
    NextPageTokenInterface,
    PaginationBase,
    create_next_page_token,
)

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,
        )

NextPageToken = create_next_page_token(Pagination, "NextPageToken")

5. `client.py`

Purpose: Handles communication with your application’s API. Key aspects:

Authentication with your API
Methods for each API endpoint
Error handling for API calls
Transformation of API responses

Example implementation:

import typing as t
from connector.oai.base_clients import BaseIntegrationClient
from connector.oai.capability import Request, get_oauth, get_settings
from connector.utils.httpx_auth import BearerAuth

from myapp_connector.constants import API_BASE_PATH
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)
        return {
            "auth": BearerAuth(token=get_oauth(args).access_token),
            "base_url": settings.api_url,
        }
    
    async def get_users(self, limit: int | None = None, offset: int | None = None):
        """Fetch users from the API with pagination."""
        params = {}
        if limit:
            params["limit"] = limit
        if offset:
            params["offset"] = offset
            
        response = await self._http_client.get(f"{API_BASE_PATH}/users", params=params)
        return response.json()
    
    # Additional methods for other API endpoints

6. `capabilities_read.py`

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

Translates between your API and Lumos data formats
Handles pagination
Implements error handling

Example implementation:

from connector.generated import (
    ListAccountsRequest,
    ListAccountsResponse,
    FoundAccountData,
    Page,
)
from connector.oai.capability import get_page

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

async def list_accounts(args: ListAccountsRequest) -> ListAccountsResponse:
    """List user accounts from the application."""
    endpoint = "/api/users"
    
    # Get pagination info or use defaults
    try:
        current_pagination = NextPageToken(get_page(args).token).paginations()[0]
    except IndexError:
        current_pagination = Pagination.default(endpoint)

    page_size = get_page(args).size or DEFAULT_PAGE_SIZE
    
    # Make API call
    async with MyAppClient(args) as client:
        response = await client.get_users(
            limit=page_size,
            offset=current_pagination.offset,
        )
        
        # Transform API response to Lumos format
        accounts = []
        for user in response.get("items", []):
            accounts.append(FoundAccountData(
                account_id=user["id"],
                email=user.get("email", ""),
                display_name=user.get("name", ""),
                status="active" if user.get("active", True) else "inactive",
            ))
        
        # Handle pagination for next page
        has_more = response.get("has_more", False)
        next_pagination = []
        if has_more:
            next_pagination.append(
                Pagination(
                    endpoint=endpoint,
                    offset=current_pagination.offset + len(accounts),
                )
            )

        next_page_token = NextPageToken.from_paginations(next_pagination).token if next_pagination else None

    return ListAccountsResponse(
        response=accounts,
        page=Page(token=next_page_token, size=page_size) if next_page_token else None,
    )

7. `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
Implements error handling

Example implementation:

from connector.generated import (
    CreateAccountRequest,
    CreateAccountResponse,
    FoundAccountData,
)

from myapp_connector.client import MyAppClient

async def create_account(args: CreateAccountRequest) -> CreateAccountResponse:
    """Create a new user account in the application."""
    req = args.request
    
    # Prepare user data for the API
    user_data = {
        "email": req.email,
        "name": req.display_name,
        "status": "active",
    }
    
    async with MyAppClient(args) as client:
        # Make API call to create user
        new_user = await client.create_user(user_data)
        
        # Return the created account in Lumos format
        return CreateAccountResponse(
            response=FoundAccountData(
                account_id=new_user["id"],
                email=new_user["email"],
                display_name=new_user.get("name", ""),
                status="active",
            )
        )

8. `integration.py`

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

Defines connector metadata (name, description, logo)
Registers capabilities
Configures error handling
Specifies authentication type

Example implementation:

import httpx
from connector.generated import OAuthCredential
from connector.oai.capability import StandardCapabilityName
from connector.oai.errors import HTTPHandler
from connector.oai.integration import DescriptionData, Integration, AppCategory

from myapp_connector.__about__ import __version__
from myapp_connector.enums import entitlement_types, resource_types
from myapp_connector.settings import MyAppSettings
from myapp_connector import capabilities_read, capabilities_write

integration = Integration(
    app_id="myapp",
    version=__version__,
    auth=OAuthCredential,
    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,
    }
)

9. `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 '{"auth":...,"request":{},"settings":...}'

Common Authentication Types

Lumos supports multiple authentication methods:

OAuthCredential - For OAuth 2.0 flows
BasicCredential - For username/password authentication
TokenCredential - For API key or token-based authentication
JWTCredential - For JWT-based authentication

Choose the appropriate type in your integration.py.

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

Authentication issues: Check token format and scope permissions
Missing data: Verify field mappings between your API and Lumos models
Pagination errors: Ensure your NextPageToken implementation is correct
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 OAuth support - If your application uses OAuth

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

Connect an App

Supported Generics

Webhooks, ITSM, and Manual Tasks

On-Premise Agent

Connector SDK

Managing Integrations

Integration Library

Quick Start Guide

Implementation Workflow

Minimum Viable Connector

Core Files Overview

1. `settings.py`

2. `constants.py`

3. `enums.py`

5. `client.py`

6. `capabilities_read.py`

7. `capabilities_write.py`

8. `integration.py`

9. `main.py`

10. `about.py`

Testing Your Connector

Common Authentication Types

Key Capabilities

Troubleshooting Tips

Next Steps

​Implementation Workflow

​Minimum Viable Connector

​Core Files Overview

​1. settings.py

​2. constants.py

​3. enums.py

​4. pagination.py

​5. client.py

​6. capabilities_read.py

​7. capabilities_write.py

​8. integration.py

​9. main.py

​10. __about__.py

​Testing Your Connector

​Common Authentication Types

​Key Capabilities

​Troubleshooting Tips

​Next Steps

Implementation Workflow

Minimum Viable Connector

Core Files Overview

1. `settings.py`

2. `constants.py`

3. `enums.py`

4. `pagination.py`

5. `client.py`

6. `capabilities_read.py`

7. `capabilities_write.py`

8. `integration.py`

9. `main.py`

10. `about.py`

Testing Your Connector

Common Authentication Types

Key Capabilities

Troubleshooting Tips

Next Steps