Building a Lumos Connector: Step-by-Step Tutorial

This tutorial guides you through the entire process of building a custom Lumos connector for your application. By the end, you'll have:

A working test server that simulates your application's API
A fully functional Lumos connector that integrates with this API
The knowledge to adapt this connector for your real application

Setup Overview

We'll be working with two components:

Mock API Server: A FastAPI application that simulates your application's API endpoints
Lumos Connector: A Python package that translates between your API and Lumos

Part 1: Setting Up the Test Environment

Step 1: Install Dependencies

# Create a project directory
mkdir lumos-connector-project
cd lumos-connector-project

# Create a virtual environment
python -m venv venv
source venv/bin/activate  # On Windows: venv\Scripts\activate

# Install required packages
pip install "connector-py[dev]" fastapi uvicorn httpx "pydantic[email]" python-multipart

Step 2: Set Up the Mock API Server

Make sure to create the Mock API Server file, follow the instructions in Mock API Server for Testing Lumos Connectors.

From a new terminal window, we'll start the mock API server:

source venv/bin/activate
uvicorn mock_api_server:app --reload

The server should now be running at http://localhost:8000. You can access the API documentation at http://localhost:8000/docs. Do not close this terminal window/session.

Step 3: Test the API Server

Let's verify that the server is working correctly, from your previous terminal window run:

# Check the health endpoint
curl http://localhost:8000/api/health

# List users (requires authentication)
curl -H "Authorization: Bearer valid-token" http://localhost:8000/api/users

Part 2: Creating the Lumos Connector

Step 1: Scaffold a New Connector

Now we'll create a new Lumos connector that will connect to our mock API:

# Create a new connector (from project root)
connector scaffold mockapp-connector mockapp_connector \
  --author-name "Your Name" --author-email "[email protected]"

# Navigate to the connector directory
cd mockapp_connector

# Install the connector dependencies
pip install -e ".[all]"

If you omit --author-name and --author-email, the command prompts for them interactively.

The scaffold generates a multi-auth connector: the credentials it accepts are declared in auth.py and passed to the Integration as credentials. By default it declares a single OAuth credential, which we will change to an API token, since that is what our mock API uses.

Step 2: Configure the Connector Settings

Edit mockapp_connector/settings.py to define the non-secret configuration users will provide. The API token is a credential, not a setting, so it does not belong here:

from pydantic import BaseModel, Field


class MockappConnectorSettings(BaseModel):
    """Settings for MockApp connector."""

    api_url: str = Field(
        default="http://localhost:8000",
        description="The base URL of the MockApp API",
    )

Step 3: Update Constants

Edit mockapp_connector/constants.py. Keep the credential ID enum that the scaffold generated - auth.py imports from it - and replace the OAuth URLs, which this connector does not use:

from enum import Enum

# Only used by the generated test suite as the mocked host. The client builds its
# real base URL from the `api_url` setting.
BASE_URL = "http://localhost:8000"
API_BASE_PATH = "/api"


class MockappConnectorCredentialId(str, Enum):
    """IDs of the credentials this connector accepts."""

    API_TOKEN = "mockapp_api_token"

Do not delete BASE_URL: tests/test_all_capabilities.py imports it as the host to mock.

Step 4: Declare the Credential

Edit mockapp_connector/auth.py. The scaffold generates an OAuth credential; replace it with a single API token credential:

"""Credential configuration for the MockApp connector."""

from connector_sdk_types import AuthModel, CredentialConfig

from mockapp_connector.constants import MockappConnectorCredentialId

API_TOKEN_DESCRIPTION = """The MockApp API token used to authenticate API calls.

1. Sign in to MockApp as an administrator
2. Open Settings -> API tokens
3. Create a token and copy it into the field below
"""

MockappCredentialsConfig = [
    CredentialConfig(
        id=MockappConnectorCredentialId.API_TOKEN,
        name="API Token",
        type=AuthModel.TOKEN,
        description=API_TOKEN_DESCRIPTION,
    ),
]

Step 5: Define Resource and Entitlement Types

Edit mockapp_connector/enums.py to match the types in our mock API:

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


class MockappResourceTypes(str, Enum):
    PROJECT = "project"
    TEAM = "team"
    GLOBAL = "global"


class MockappEntitlementTypes(str, Enum):
    PROJECT_ROLE = "project_role"
    TEAM_MEMBER = "team_member"
    ADMIN_ROLE = "admin_role"


resource_types: list[ResourceType] = [
    ResourceType(type_id=MockappResourceTypes.PROJECT, type_label="Project"),
    ResourceType(type_id=MockappResourceTypes.TEAM, type_label="Team"),
    ResourceType(type_id=MockappResourceTypes.GLOBAL, type_label="Global Resource"),
]

entitlement_types: list[EntitlementType] = [
    EntitlementType(
        type_id=MockappEntitlementTypes.PROJECT_ROLE,
        type_label="Project Role",
        resource_type_id=MockappResourceTypes.PROJECT,
        min=0,
    ),
    EntitlementType(
        type_id=MockappEntitlementTypes.TEAM_MEMBER,
        type_label="Team Member",
        resource_type_id=MockappResourceTypes.TEAM,
        min=0,
    ),
    EntitlementType(
        type_id=MockappEntitlementTypes.ADMIN_ROLE,
        type_label="Admin Role",
        resource_type_id=MockappResourceTypes.GLOBAL,
        min=0,
        max=1,  # Users can have at most one admin role
    ),
]

min is required on EntitlementType. Add max only where the app genuinely caps assignments.

Step 6: Create API Client

Edit mockapp_connector/client.py. Two things matter here: get_credential reads the credential by the ID declared in auth.py, and every method calls response.raise_for_status():

import typing as t

from connector.generated import TokenCredential
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 mockapp_connector.constants import API_BASE_PATH, MockappConnectorCredentialId
from mockapp_connector.settings import MockappConnectorSettings


class MockappClient(BaseIntegrationClient):
    @classmethod
    def prepare_client_args(cls, args: Request) -> dict[str, t.Any]:
        settings = get_settings(args, MockappConnectorSettings)
        token = get_credential(args, MockappConnectorCredentialId.API_TOKEN, TokenCredential)
        return {
            "auth": BearerAuth(token=token.token),
            "base_url": settings.api_url,
        }

    async def get_users(
        self, limit: int | None = None, offset: int | None = None
    ) -> dict[str, t.Any]:
        """Fetch a page of users from the API."""
        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())

    async def get_user(self, user_id: str) -> dict[str, t.Any]:
        """Fetch a single user."""
        response = await self._http_client.get(f"{API_BASE_PATH}/users/USER_ID")
        response.raise_for_status()
        return t.cast(dict[str, t.Any], response.json())

    async def get_resources(self) -> list[dict[str, t.Any]]:
        response = await self._http_client.get(f"{API_BASE_PATH}/resources")
        response.raise_for_status()
        return t.cast(list[dict[str, t.Any]], response.json())

    async def get_entitlements(self) -> list[dict[str, t.Any]]:
        response = await self._http_client.get(f"{API_BASE_PATH}/entitlements")
        response.raise_for_status()
        return t.cast(list[dict[str, t.Any]], response.json())

    async def get_associations(self) -> list[dict[str, t.Any]]:
        response = await self._http_client.get(f"{API_BASE_PATH}/associations")
        response.raise_for_status()
        return t.cast(list[dict[str, t.Any]], response.json())

    async def create_user(self, user_data: dict[str, t.Any]) -> dict[str, t.Any]:
        response = await self._http_client.post(f"{API_BASE_PATH}/users", json=user_data)
        response.raise_for_status()
        return t.cast(dict[str, t.Any], response.json())

    async def update_user(
        self, user_id: str, user_data: dict[str, t.Any]
    ) -> dict[str, t.Any]:
        response = await self._http_client.put(
            f"{API_BASE_PATH}/users/USER_ID", json=user_data
        )
        response.raise_for_status()
        return t.cast(dict[str, t.Any], response.json())

    async def delete_user(self, user_id: str) -> None:
        response = await self._http_client.delete(f"{API_BASE_PATH}/users/USER_ID")
        response.raise_for_status()

    async def assign_entitlement(
        self, user_id: str, entitlement_id: str, resource_id: str
    ) -> None:
        response = await self._http_client.post(
            f"{API_BASE_PATH}/users/USER_ID/entitlements",
            json={"entitlement_id": entitlement_id, "resource_id": resource_id},
        )
        response.raise_for_status()

    async def unassign_entitlement(
        self, user_id: str, entitlement_id: str, resource_id: str
    ) -> None:
        response = await self._http_client.delete(
            f"{API_BASE_PATH}/users/USER_ID/entitlements/{entitlement_id}",
            params={"resource_id": resource_id},
        )
        response.raise_for_status()

Why raise_for_status() matters. An HTTP 401 from this API still returns a JSON body, so without it response.json() succeeds and the failure looks like a success. Your validate_credentials would then report an invalid token as "valid": true.

Step 7: Implement Read Capabilities

Edit mockapp_connector/capabilities_read.py:

import httpx
from connector.generated import (
    AccountStatus,
    ActivityEventType,
    FindEntitlementAssociationsRequest,
    FindEntitlementAssociationsResponse,
    FoundAccountData,
    FoundEntitlementAssociation,
    FoundEntitlementData,
    FoundResourceData,
    GetLastActivityRequest,
    GetLastActivityResponse,
    LastActivityData,
    ListAccountsRequest,
    ListAccountsResponse,
    ListEntitlementsRequest,
    ListEntitlementsResponse,
    ListResourcesRequest,
    ListResourcesResponse,
    ValidateCredentialsRequest,
    ValidateCredentialsResponse,
    ValidatedCredentials,
)

from mockapp_connector.client import MockappClient
from mockapp_connector.constants import API_BASE_PATH
from mockapp_connector.pagination import NextPageToken, Pagination, paginations_from_args

USERS_ENDPOINT = f"{API_BASE_PATH}/users"

# MockApp reports user state as a plain string; map it onto Lumos account statuses.
ACCOUNT_STATUS_MAP = {
    "active": AccountStatus.ACTIVE,
    "inactive": AccountStatus.INACTIVE,
    "deleted": AccountStatus.DELETED,
}


async def validate_credentials(args: ValidateCredentialsRequest) -> ValidateCredentialsResponse:
    """Prove the credentials work by making one cheap authenticated call."""
    async with MockappClient(args) as client:
        await client.get_users(limit=1)

    return ValidateCredentialsResponse(
        response=ValidatedCredentials(
            unique_tenant_id="mockapp-tenant-id",
            valid=True,
        ),
    )


async def list_accounts(args: ListAccountsRequest) -> ListAccountsResponse:
    """List user accounts from MockApp, one page at a time."""
    paginations, current_pagination, page_size = paginations_from_args(
        args, default_endpoints=[USERS_ENDPOINT]
    )

    async with MockappClient(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=ACCOUNT_STATUS_MAP.get(
                    user.get("status", ""), 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,
    )


async def list_resources(args: ListResourcesRequest) -> ListResourcesResponse:
    """List resources from MockApp."""
    async with MockappClient(args) as client:
        resources_data = await client.get_resources()

    return ListResourcesResponse(
        response=[
            FoundResourceData(
                integration_specific_id=resource["id"],
                resource_type=resource["type"],
                label=resource["name"],
            )
            for resource in resources_data
        ],
    )


async def list_entitlements(args: ListEntitlementsRequest) -> ListEntitlementsResponse:
    """List entitlements from MockApp."""
    async with MockappClient(args) as client:
        entitlements_data = await client.get_entitlements()

    return ListEntitlementsResponse(
        response=[
            FoundEntitlementData(
                integration_specific_id=entitlement["id"],
                entitlement_type=entitlement["type"],
                integration_specific_resource_id=entitlement["resource_id"],
                label=entitlement["name"],
            )
            for entitlement in entitlements_data
        ],
    )


async def find_entitlement_associations(
    args: FindEntitlementAssociationsRequest,
) -> FindEntitlementAssociationsResponse:
    """Find which accounts hold which entitlements in MockApp."""
    async with MockappClient(args) as client:
        associations_data = await client.get_associations()

    return FindEntitlementAssociationsResponse(
        response=[
            FoundEntitlementAssociation(
                account_id=assoc["user_id"],
                integration_specific_entitlement_id=assoc["entitlement_id"],
                integration_specific_resource_id=assoc["resource_id"],
            )
            for assoc in associations_data
        ],
    )


async def get_last_activity(args: GetLastActivityRequest) -> GetLastActivityResponse:
    """Report the last login for each requested account."""
    activities: list[LastActivityData] = []

    async with MockappClient(args) as client:
        for account_id in args.request.account_ids:
            try:
                user = await client.get_user(account_id)
            except httpx.HTTPStatusError as exc:
                if exc.response.status_code == httpx.codes.NOT_FOUND:
                    continue  # Account no longer exists in MockApp; skip it.
                raise

            if user.get("last_login"):
                activities.append(
                    LastActivityData(
                        account_id=account_id,
                        event_type=ActivityEventType.LAST_LOGIN,
                        happened_at=user["last_login"],
                    )
                )

    return GetLastActivityResponse(response=activities)

Two patterns worth noticing:

There are no broad try/except blocks. The HTTPStatusError handler registered in integration.py converts API failures into typed connector errors with the right error code, which is more useful than re-wrapping them.
get_last_activity catches only the specific 404 it wants to tolerate, and re-raises everything else.

Step 8: Implement Write Capabilities

Edit mockapp_connector/capabilities_write.py:

from connector.generated import (
    AccountStatus,
    ActivateAccountRequest,
    ActivateAccountResponse,
    ActivatedAccount,
    AssignEntitlementRequest,
    AssignEntitlementResponse,
    AssignedEntitlement,
    CreateAccountResponse,
    CreatedAccount,
    DeactivateAccountRequest,
    DeactivateAccountResponse,
    DeactivatedAccount,
    DeleteAccountRequest,
    DeleteAccountResponse,
    DeletedAccount,
    UnassignedEntitlement,
    UnassignEntitlementRequest,
    UnassignEntitlementResponse,
)
from connector.oai.capability import CustomRequest

from mockapp_connector.client import MockappClient
from mockapp_connector.dto.user import CreateAccount


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

    async with MockappClient(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,
        )
    )


async def activate_account(args: ActivateAccountRequest) -> ActivateAccountResponse:
    """Activate a user account in MockApp."""
    async with MockappClient(args) as client:
        await client.update_user(args.request.account_id, {"status": "active"})

    return ActivateAccountResponse(
        response=ActivatedAccount(activated=True, status=AccountStatus.ACTIVE)
    )


async def deactivate_account(args: DeactivateAccountRequest) -> DeactivateAccountResponse:
    """Deactivate a user account in MockApp."""
    async with MockappClient(args) as client:
        await client.update_user(args.request.account_id, {"status": "inactive"})

    return DeactivateAccountResponse(
        response=DeactivatedAccount(deactivated=True, status=AccountStatus.INACTIVE)
    )


async def delete_account(args: DeleteAccountRequest) -> DeleteAccountResponse:
    """Delete a user account in MockApp."""
    async with MockappClient(args) as client:
        await client.delete_user(args.request.account_id)

    return DeleteAccountResponse(
        response=DeletedAccount(deleted=True, status=AccountStatus.DELETED)
    )


async def assign_entitlement(args: AssignEntitlementRequest) -> AssignEntitlementResponse:
    """Assign an entitlement to a user in MockApp."""
    request = args.request

    async with MockappClient(args) as client:
        await client.assign_entitlement(
            request.account_integration_specific_id,
            request.entitlement_integration_specific_id,
            request.resource_integration_specific_id,
        )

    return AssignEntitlementResponse(response=AssignedEntitlement(assigned=True))


async def unassign_entitlement(
    args: UnassignEntitlementRequest,
) -> UnassignEntitlementResponse:
    """Unassign an entitlement from a user in MockApp."""
    request = args.request

    async with MockappClient(args) as client:
        await client.unassign_entitlement(
            request.account_integration_specific_id,
            request.entitlement_integration_specific_id,
            request.resource_integration_specific_id,
        )

    return UnassignEntitlementResponse(response=UnassignedEntitlement(unassigned=True))

Step 9: Update the User DTO

create_account receives whatever fields your app needs to create a user, declared in mockapp_connector/dto/user.py. MockApp needs an email and a display name:

from connector.generated import CreatableAccount
from pydantic import Field


# TODO: add/remove fields required to create account
class CreateAccount(CreatableAccount):
    email: str = Field(description="The email address for the new account")
    display_name: str = Field(description="The display name for the new account")

Step 10: Configure the Integration

Edit mockapp_connector/integration.py to pass the credential list and register all capabilities:

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 mockapp_connector import capabilities_read, capabilities_write
from mockapp_connector.__about__ import __version__
from mockapp_connector.auth import MockappCredentialsConfig
from mockapp_connector.constants import MockappConnectorCredentialId
from mockapp_connector.enums import entitlement_types, resource_types
from mockapp_connector.settings import MockappConnectorSettings

integration = Integration(
    app_id="mockapp-connector",
    version=__version__,
    credentials=MockappCredentialsConfig,
    credentials_settings=CredentialsSettings(
        allowed_credentials=[
            (MockappConnectorCredentialId.API_TOKEN,),
        ],
    ),
    exception_handlers=[
        (httpx.HTTPStatusError, HTTPHandler, None),
    ],
    description_data=DescriptionData(
        logo_url="https://logo.clearbit.com/example.com",
        user_friendly_name="MockApp",
        description="MockApp is a test application for demonstrating Lumos connectors",
        categories=[AppCategory.DEVELOPERS, AppCategory.IT_AND_SECURITY],
    ),
    settings_model=MockappConnectorSettings,
    resource_types=resource_types,
    entitlement_types=entitlement_types,
)

# Register all 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,
        StandardCapabilityName.FIND_ENTITLEMENT_ASSOCIATIONS: capabilities_read.find_entitlement_associations,
        StandardCapabilityName.GET_LAST_ACTIVITY: capabilities_read.get_last_activity,
        # Write capabilities
        StandardCapabilityName.CREATE_ACCOUNT: capabilities_write.create_account,
        StandardCapabilityName.ACTIVATE_ACCOUNT: capabilities_write.activate_account,
        StandardCapabilityName.DEACTIVATE_ACCOUNT: capabilities_write.deactivate_account,
        StandardCapabilityName.DELETE_ACCOUNT: capabilities_write.delete_account,
        StandardCapabilityName.ASSIGN_ENTITLEMENT: capabilities_write.assign_entitlement,
        StandardCapabilityName.UNASSIGN_ENTITLEMENT: capabilities_write.unassign_entitlement,
    }
)

Keep app_id as the scaffold generated it (mockapp-connector). It is used to locate your module when gathering test cases, so changing it can break the test runner.

validate_credential_config is not listed here - the SDK registers it automatically for multi-auth connectors.

Step 11: Update the Generated Test Fixtures

The scaffold generated test fixtures for the default OAuth connector. Because we changed the credential to a token and edited the account DTO, three of those files no longer match and mypy . will fail until you update them.

Replace tests/common_mock_data.py:

from datetime import datetime, timezone

from connector.generated import AuthCredential, TokenCredential
from mockapp_connector.constants import MockappConnectorCredentialId
from mockapp_connector.settings import MockappConnectorSettings

VALID_AUTH = [
    AuthCredential(
        id=MockappConnectorCredentialId.API_TOKEN,
        token=TokenCredential(token="valid-token"),  # noqa: S106
    ),
]
INVALID_AUTH = [
    AuthCredential(
        id=MockappConnectorCredentialId.API_TOKEN,
        token=TokenCredential(token="invalid-token"),  # noqa: S106
    ),
]
SETTINGS = MockappConnectorSettings(api_url="http://localhost:8000").model_dump()
TEST_MAX_PAGE_SIZE = 100

DATETIME_NOW = datetime.now(tz=timezone.utc)

Replace tests/test_read_capabilities/test_validate_credential_config_cases.py:

"""Cases for testing the ``validate_credential_config`` capability."""

import typing as t

from connector.generated import AuthCredential, StandardCapabilityName, TokenCredential
from connector.tests.type_definitions import ResponseBodyMap
from connector_sdk_types import (
    ValidateCredentialConfig,
    ValidateCredentialConfigRequest,
    ValidateCredentialConfigResponse,
    ValidatedCredentialConfig,
)
from mockapp_connector.constants import MockappConnectorCredentialId

from tests.common_mock_data import SETTINGS

Case: t.TypeAlias = tuple[
    StandardCapabilityName,
    ValidateCredentialConfigRequest,
    ResponseBodyMap,
    ValidateCredentialConfigResponse,
]


def case_validate_credential_config_200() -> Case:
    """A well-formed API token passes validation."""
    args = ValidateCredentialConfigRequest(
        request=ValidateCredentialConfig(
            credential=AuthCredential(
                id=MockappConnectorCredentialId.API_TOKEN,
                token=TokenCredential(token="valid-token"),  # noqa: S106
            ),
        ),
        settings=SETTINGS,
    )
    response_body_map: ResponseBodyMap = {}
    expected_response = ValidateCredentialConfigResponse(
        response=ValidatedCredentialConfig(valid=True, validation_errors=[]),
    )
    return (
        StandardCapabilityName.VALIDATE_CREDENTIAL_CONFIG,
        args,
        response_body_map,
        expected_response,
    )


def case_validate_credential_config_empty_token() -> Case:
    """The SDK's base validation rejects an empty required field."""
    args = ValidateCredentialConfigRequest(
        request=ValidateCredentialConfig(
            credential=AuthCredential(
                id=MockappConnectorCredentialId.API_TOKEN,
                token=TokenCredential(token=""),
            ),
        ),
        settings=SETTINGS,
    )
    response_body_map: ResponseBodyMap = {}
    expected_response = ValidateCredentialConfigResponse(
        response=ValidatedCredentialConfig(
            valid=False,
            validation_errors=[
                "A required field is empty, please provide a value before submitting again"
            ],
        ),
    )
    return (
        StandardCapabilityName.VALIDATE_CREDENTIAL_CONFIG,
        args,
        response_body_map,
        expected_response,
    )

Finally, in tests/test_write_capabilities/test_create_account_cases.py, add a display_name to each of the three CreateAccount(...) calls, since we made that field required:

    request = CreateAccount(
        email="[email protected]",
        display_name="John Doe",
        entitlements=[],
    )

Part 3: Testing Your Connector

Step 1: Typecheck

From the connector's root directory:

# Run the type checker
mypy .

You should see:

Success: no issues found in 32 source files

If you instead get a wall of Cannot find implementation or library stub for module named "connector...", the SDK is not installed in the environment you are running mypy from.

Re-run pip install -e ".[all]" inside your activated virtual environment.

Step 2: Understand the Generated Tests

pytest tests

At this point most tests fail, and that is expected. The scaffold generates a case file per capability, but the cases are placeholders that mock a fictional GET /example endpoint. Now that the capabilities call real MockApp endpoints, each placeholder fails with:

AssertionError: The following responses are mocked but not requested:
  - Match GET request on http://localhost:8000/example

You replace them one at a time. Here is a complete, working replacement for tests/test_read_capabilities/test_list_accounts_cases.py to use as your pattern:

"""Cases for testing the ``list_accounts`` capability."""

import typing as t

import httpx
from connector.generated import (
    AccountStatus,
    ErrorResponse,
    FoundAccountData,
    ListAccounts,
    ListAccountsRequest,
    ListAccountsResponse,
    Page,
    StandardCapabilityName,
)
from connector.tests.type_definitions import MockedResponse, ResponseBodyMap
from mockapp_connector.pagination import NextPageToken, Pagination

from tests.common_mock_data import SETTINGS, VALID_AUTH

Case: t.TypeAlias = tuple[
    StandardCapabilityName,
    ListAccountsRequest,
    ResponseBodyMap,
    ListAccountsResponse | ErrorResponse,
]

TEST_PAGE_SIZE = 2


def case_list_accounts_200() -> Case:
    """First page, with more results available."""
    args = ListAccountsRequest(
        request=ListAccounts(),
        credentials=VALID_AUTH,
        settings=SETTINGS,
        page=Page(size=TEST_PAGE_SIZE),
    )
    response_body_map: ResponseBodyMap = {
        "GET": {
            "/api/users?limit=2&offset=0": MockedResponse(
                status_code=httpx.codes.OK,
                response_body={
                    "items": [
                        {
                            "id": "user-1",
                            "email": "[email protected]",
                            "name": "User One",
                            "status": "active",
                        },
                        {
                            "id": "user-2",
                            "email": "[email protected]",
                            "name": "User Two",
                            "status": "inactive",
                        },
                    ],
                    "total": 3,
                    "has_more": True,
                },
            ),
        },
    }
    expected_response = ListAccountsResponse(
        response=[
            FoundAccountData(
                integration_specific_id="user-1",
                email="[email protected]",
                username="[email protected]",
                given_name="User One",
                user_status=AccountStatus.ACTIVE,
            ),
            FoundAccountData(
                integration_specific_id="user-2",
                email="[email protected]",
                username="[email protected]",
                given_name="User Two",
                user_status=AccountStatus.INACTIVE,
            ),
        ],
        page=NextPageToken.from_paginations(
            [Pagination(endpoint="/api/users", offset=2)]
        ).to_page(size=TEST_PAGE_SIZE),
    )
    return StandardCapabilityName.LIST_ACCOUNTS, args, response_body_map, expected_response


def case_list_accounts_200_last_page() -> Case:
    """Final page: no next page token is returned."""
    args = ListAccountsRequest(
        request=ListAccounts(),
        credentials=VALID_AUTH,
        settings=SETTINGS,
        page=Page(size=TEST_PAGE_SIZE),
    )
    response_body_map: ResponseBodyMap = {
        "GET": {
            "/api/users?limit=2&offset=0": MockedResponse(
                status_code=httpx.codes.OK,
                response_body={"items": [], "total": 0, "has_more": False},
            ),
        },
    }
    expected_response = ListAccountsResponse(response=[], page=None)
    return StandardCapabilityName.LIST_ACCOUNTS, args, response_body_map, expected_response

Verify just the files you have rewritten:

pytest tests -k "list_accounts or validate_credential_config"

....                                                     [100%]
4 passed, 17 deselected

Points to carry over to the other case files:

Credentials go on the request's credentials field as a list, not auth.
Mocked URLs are matched against the host in BASE_URL, including the query string.
Case files must be named test_{capability_name}_cases.py, or the runner will not find them.
Every registered capability needs a case file.

Step 3: Test Against the Live Mock API

With the mock API server still running, exercise the connector for real. Credentials are passed as a credentials array, tagged with the ID from auth.py:

# Validate credentials
mockapp-connector validate_credentials --json '{"credentials":[{"id":"mockapp_api_token","token":{"token":"valid-token"}}],"request":{},"settings":{"api_url":"http://localhost:8000"}}'

{"response":{"valid":true,"unique_tenant_id":"mockapp-tenant-id"},"raw_data":null,"page":null,"rate_limit":null,"execution_summary":null}

# List accounts, two at a time
mockapp-connector list_accounts --json '{"credentials":[{"id":"mockapp_api_token","token":{"token":"valid-token"}}],"request":{},"settings":{"api_url":"http://localhost:8000"},"page":{"size":2}}'

The response includes a page.token. Feed it back to fetch the next page:

mockapp-connector list_accounts --json '{"credentials":[{"id":"mockapp_api_token","token":{"token":"valid-token"}}],"request":{},"settings":{"api_url":"http://localhost:8000"},"page":{"size":2,"token":"<token from the previous response>"}}'

Confirm that a bad token is reported as a failure rather than a success:

mockapp-connector validate_credentials --json '{"credentials":[{"id":"mockapp_api_token","token":{"token":"bogus"}}],"request":{},"settings":{"api_url":"http://localhost:8000"}}'

{"is_error":true,"error":{"message":"[401][http://localhost:8000/api/users] {'detail': 'Invalid token'}","status_code":401,"error_code":"unauthorized","app_id":"mockapp-connector","raised_by":"HTTPStatusError","raised_in":"mockapp_connector.capabilities_read:validate_credentials", ...}}

Step 4: Test the Remaining Capabilities

For brevity, $CRED below stands for "credentials":[{"id":"mockapp_api_token","token":{"token":"valid-token"}}] and $S for "settings":{"api_url":"http://localhost:8000"}.

mockapp-connector list_resources --json '{'$CRED',"request":{},'$S'}'
mockapp-connector list_entitlements --json '{'$CRED',"request":{},'$S'}'
mockapp-connector find_entitlement_associations --json '{'$CRED',"request":{},'$S'}'
mockapp-connector get_last_activity --json '{'$CRED',"request":{"account_ids":["user-1","user-3"]},'$S'}'

mockapp-connector create_account --json '{'$CRED',"request":{"email":"[email protected]","display_name":"New User","entitlements":[]},'$S'}'
mockapp-connector deactivate_account --json '{'$CRED',"request":{"account_id":"user-4"},'$S'}'
mockapp-connector activate_account --json '{'$CRED',"request":{"account_id":"user-4"},'$S'}'
mockapp-connector delete_account --json '{'$CRED',"request":{"account_id":"user-6"},'$S'}'

mockapp-connector assign_entitlement --json '{'$CRED',"request":{"account_integration_specific_id":"user-5","entitlement_integration_specific_id":"team-lead","resource_integration_specific_id":"team-1","resource_type":"team","entitlement_type":"team_member"},'$S'}'
mockapp-connector unassign_entitlement --json '{'$CRED',"request":{"account_integration_specific_id":"user-5","entitlement_integration_specific_id":"team-lead","resource_integration_specific_id":"team-1","resource_type":"team","entitlement_type":"team_member"},'$S'}'

Each returns a small success payload, for example:

{"response":{"assigned":true},"raw_data":null,"page":null,"rate_limit":null,"execution_summary":null}
{"response":{"status":"DELETED","deleted":true},"raw_data":null,"page":null,"rate_limit":null,"execution_summary":null}

get_last_activity returns entries only for accounts that have a last_login, and silently skips accounts the API no longer knows about.

Part 4: Adapting for Your Real Application

To adapt this connector for your real application, you'll need to:

Update the settings: modify settings.py for any non-secret configuration your app needs
Change the credentials: update auth.py to declare the credentials your app requires, and add an ID per credential to constants.py
Adjust the client: update client.py to match your API's endpoints and request/response formats, using one client class per credential
Map data models: update the transformation logic in capabilities_read.py and capabilities_write.py

Key Files and Their Purpose

Here's a summary of the key files and what they're responsible for:

File	Purpose
integration.py	Configures the connector and registers capabilities
auth.py	Declares the credentials the connector accepts
settings.py	Defines the non-secret configuration settings
constants.py	Contains API URLs and the credential ID enum
enums.py	Defines resource and entitlement types
client.py	Implements API communication
capabilities_read.py	Implements read operations
capabilities_write.py	Implements write operations
pagination.py	Handles pagination for large data sets
dto/*.py	Declares the necessary inputs/outputs for the API objects

Common Patterns and Best Practices

Always raise_for_status(): otherwise error responses parse as successful ones
Let handlers handle errors: register exception handlers rather than wrapping every call in try/except
Catch narrowly: only swallow the specific status code you mean to tolerate, and re-raise the rest
Data transformation: create helper functions for transforming between API and Lumos data models
Pagination: follow the paginations_from_args pattern shown in list_accounts
Authentication: keep credential lookup in the client's prepare_client_args method
Testing: replace the generated placeholder cases as you implement each capability

Troubleshooting

Symptom	Cause
`ModuleNotFoundError: No module named 'PyInstaller'`	Install the SDK with the `[dev]` extra
`mypy` reports dozens of `Cannot find implementation or library stub`	SDK not installed in the environment running mypy
`validate_credentials` returns `valid: true` with a bad token	Missing `response.raise_for_status()` in the client
`Module "...constants" has no attribute "..."`	You renamed the credential ID enum but a test fixture still uses the old name
`The following responses are mocked but not requested`	A placeholder case file still mocks `/example`
`Missing named argument "display_name"`	The DTO gained a required field the create_account cases don't pass
Association/entitlement calls 404	Check that `resource_id` and `entitlement_id` exist in the mock data

Next Steps

Implement Tests: replace the remaining placeholder case files
Add Documentation: document any unique behavior or requirements
Deploy: use the packaging tools to deploy your connector
Monitor: add logging to help troubleshoot issues in production

Conclusion

You now have a working Lumos connector that can integrate with your application. By following the patterns in this tutorial, you can extend it to support additional capabilities or adapt it for other applications.