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?
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.

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

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`