---
updatedAt: 2026-08-04T14:30:10.000Z
agentTools:
  projectIndex: https://developers.lumos.com/llms.txt
---

# 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:

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

## Setup Overview

We'll be working with two components:

1. [Mock API Server](https://developers.lumos.com/docs/mock-api-server-for-testing-lumos-connectors#/): A FastAPI application that simulates your application's API endpoints
2. Lumos Connector: A Python package that translates between your API and Lumos

## Part 1: Setting Up the Test Environment

### Step 1: Install Dependencies

```shell
# 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](https://developers.lumos.com/docs/mock-api-server-for-testing-lumos-connectors).

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

```shell
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:

```shell
# 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:

```shell
# Create a new connector (from project root)
connector scaffold mockapp-connector mockapp_connector \
  --author-name "Your Name" --author-email "you@example.com"

# 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:

```python
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:

```python
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:

```python
"""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:

```python
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()`:

```python
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&#x20;**`raise_for_status()`**&#x20;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`:

```python
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`:

```python
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:

```python
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:

```python
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`:

```python
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`:

```python
"""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:

```python
    request = CreateAccount(
        email="jw7rT@example.com",
        display_name="John Doe",
        entitlements=[],
    )
```

## Part 3: Testing Your Connector

### Step 1: Typecheck

From the connector's root directory:

```shell
# 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.&#x20;

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

### Step 2: Understand the Generated Tests

```shell
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:

```python
"""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": "user1@example.com",
                            "name": "User One",
                            "status": "active",
                        },
                        {
                            "id": "user-2",
                            "email": "user2@example.com",
                            "name": "User Two",
                            "status": "inactive",
                        },
                    ],
                    "total": 3,
                    "has_more": True,
                },
            ),
        },
    }
    expected_response = ListAccountsResponse(
        response=[
            FoundAccountData(
                integration_specific_id="user-1",
                email="user1@example.com",
                username="user1@example.com",
                given_name="User One",
                user_status=AccountStatus.ACTIVE,
            ),
            FoundAccountData(
                integration_specific_id="user-2",
                email="user2@example.com",
                username="user2@example.com",
                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:

```shell
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`:

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

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

```shell
# 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:

```shell
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:

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

```json
{"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"}`.

```shell
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":"new.user@example.com","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:

```json
{"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:

1. **Update the settings**: modify `settings.py` for any non-secret configuration your app needs
2. **Change the credentials**: update `auth.py` to declare the credentials your app requires, and add an ID per credential to `constants.py`
3. **Adjust the client**: update `client.py` to match your API's endpoints and request/response formats, using one client class per credential
4. **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

1. **Always&#x20;**`raise_for_status()`: otherwise error responses parse as successful ones
2. **Let handlers handle errors**: register exception handlers rather than wrapping every call in `try`/`except`
3. **Catch narrowly**: only swallow the specific status code you mean to tolerate, and re-raise the rest
4. **Data transformation**: create helper functions for transforming between API and Lumos data models
5. **Pagination**: follow the `paginations_from_args` pattern shown in `list_accounts`
6. **Authentication**: keep credential lookup in the client's `prepare_client_args` method
7. **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

1. **Implement Tests**: replace the remaining placeholder case files
2. **Add Documentation**: document any unique behavior or requirements
3. **Deploy**: use the packaging tools to deploy your connector
4. **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.