scaffold-app

Bootstrap a new Atlan Application SDK connector repo from scratch. Run this first when creating a brand-new connector — before running the contract skill.

What Gets Generated

{app_name}/
├── pyproject.toml             # uv project with atlan-application-sdk dep
├── atlan.yaml                 # App manifest (app_id, execution_mode, dapr config)
├── Dockerfile                 # SDK base image + uv sync + ENV ATLAN_APP_MODULE
├── .env.example               # All required env vars with example values
├── run_dev.py                 # Local dev script using run_dev_combined()
└── app/
    ├── __init__.py
    ├── contracts.py           # Input/Output models + credential model
    ├── handler.py             # Handler subclass (test_auth, preflight, metadata)
    ├── connector.py           # App subclass with @task methods
    └── clients.py             # BaseSQLClient subclass (SQL connectors only)

Steps

1. Gather inputs

Ask the user for:

• app_name (kebab-case): e.g. my-postgres-connector
• connector_type: sql or rest
• auth_type: basic, api_key, bearer, or oauth_client

Infer class names: MyPostgresConnector, MyPostgresHandler, MyPostgresClient from kebab-case.

2. Generate pyproject.toml

[project]
name = "{app_name}"
version = "1.0.0"
requires-python = ">=3.11"
dependencies = [
    "atlan-application-sdk>=3.0.0",
    "poethepoet>=0.34.0",
]

[project.optional-dependencies]
dev = ["pytest", "pytest-asyncio"]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

[tool.uv]
dev-dependencies = [
    "pytest>=8.0",
    "pytest-asyncio>=0.23",
]

[tool.poe.tasks]
download-components.shell = """
python -c "
import application_sdk, pathlib, shutil
src = pathlib.Path(application_sdk.__file__).parent / 'components'
shutil.copytree(src, 'components', dirs_exist_ok=True)
"
"""

download-components copies the Dapr component YAMLs out of the installed atlan-application-sdk wheel — never curl them from raw.githubusercontent.com or the GitHub API; that pattern is blocked by conformance rule D009 (see docs/standards/build-security.md). Run it after uv sync and before starting daprd locally or in the Docker build. poethepoet (which provides the poe CLI) must be a main dependency, not a dev-only one — the Dockerfile's uv sync --no-dev would otherwise leave poe unavailable during the build.

3. Generate atlan.yaml

app_id: {app_name}
execution_mode: native
splitDeploymentEnabled: true
dapr:
  objectstore:
    enabled: true
  secretstore:
    enabled: true

4. Generate Dockerfile

FROM registry.atlan.com/public/app-runtime-base:3

WORKDIR /app

COPY pyproject.toml uv.lock* ./
RUN uv sync --no-dev --frozen
RUN uv run poe download-components

COPY app/ app/

ENV ATLAN_APP_MODULE=app.connector:{ClassName}App
ENV ATLAN_CONTRACT_GENERATED_DIR=app/generated

5. Generate app/contracts.py

Imports from application_sdk.contracts. Include:

• {Name}Input(Input) with connection_id: str and credential_guid: str
• {Name}Output(Output) with record_count: int
• Credential model matching the selected auth_type

6. Generate app/handler.py

Subclass Handler with test_auth, preflight_check, fetch_metadata. Use the appropriate credential type from contracts.py. Return typed AuthOutput, PreflightOutput, MetadataOutput.

For SQL connectors, fetch_metadata should use the SQL client to list databases/schemas.

7. Generate app/connector.py (SQL variant)

from application_sdk.templates import SqlMetadataExtractor
from application_sdk.templates.contracts.sql_metadata import (
    ExtractionInput, ExtractionOutput,
    FetchDatabasesInput, FetchDatabasesOutput,
)
from application_sdk.app import task

class {Name}App(SqlMetadataExtractor):
    sql_client_class = {Name}Client

    @task(timeout_seconds=1800)
    async def fetch_databases(self, input: FetchDatabasesInput) -> FetchDatabasesOutput:
        client = await self._load_sql_client(input)
        async for batch in client.run_query(self.fetch_database_sql):
            # process batch
            pass
        return FetchDatabasesOutput(chunk_count=1, total_record_count=10)

The SqlMetadataExtractor base run() already calls App.upload() to hand off artifacts to Atlan. Do not override run() without also calling super().run(input) or explicitly calling await self.upload(...) — omitting the upload is a silent failure in SDR deployments.

For REST connectors, subclass App directly with custom @task methods and an explicit App.upload() call in run():

from application_sdk.app import App, task
from application_sdk.contracts import UploadInput

class {Name}App(App):
    @task(timeout_seconds=3600)
    async def fetch_entities(self, input: {Name}Input) -> {Name}Output:
        # ... fetch and write results to input.output_path ...
        return {Name}Output(output_path=input.output_path, record_count=count)

    async def run(self, input: {Name}Input) -> {Name}Output:
        fetch_out = await self.fetch_entities(input)

        # Required: push output to Atlan's upstream store (atlan-objectstore in SDR)
        # so the publish app can index it. The activity interceptor only writes
        # FileReferences to the customer-owned objectstore. Omitting this call
        # produces a silent failure in SDR — the DAG succeeds but nothing is published.
        # See docs/concepts/file-reference.md and ADR-0014.
        await self.upload(UploadInput(local_path=fetch_out.output_path))
        return fetch_out

8. Generate app/clients.py (SQL only)

from application_sdk.clients.sql import BaseSQLClient
from application_sdk.clients.models import DatabaseConfig

class {Name}Client(BaseSQLClient):
    def _make_connection_string(self, config: DatabaseConfig) -> str:
        return f"postgresql+asyncpg://{config.username}:{config.password}@{config.host}:{config.port}/{config.database}"

9. Generate .env.example

Include all required env vars with placeholder values:

ATLAN_APP_MODULE=app.connector:{Name}App
ATLAN_TEMPORAL_HOST=localhost:7233
DAPR_HTTP_PORT=3500
DAPR_GRPC_PORT=50001
ATLAN_LOG_LEVEL=DEBUG

10. Generate run_dev.py

import asyncio
from application_sdk.main import run_dev_combined
from app.connector import {Name}App

asyncio.run(
    run_dev_combined(
        {Name}App,
        credentials={
            "host": "localhost",
            "port": "5432",
            "authType": "basic",
            "username": "admin",
            "password": "secret",
            "extra": {"database": "mydb"},
        },
        example_input={
            "connection": {
                "connection_name": "test-connection",
                "connection_qualified_name": "default/{app_name}/1234567890",
            },
        },
    )
)

After Scaffolding

Tell the user:

1. Run uv sync to install dependencies.
2. Run uv run poe download-components to copy the Dapr component YAMLs out of the installed SDK wheel into ./components (needed to run against external Dapr; the embedded runtime in step 4 doesn't require this).
3. Copy and configure .env.example → .env.
4. Run uv run python run_dev.py to test the scaffold — this boots the embedded Dapr (daprd) + in-process Temporal automatically; no Dapr CLI needed. To instead mirror production against external services, see the optional external-infrastructure section of Getting Started.
5. Run the contract skill to generate the PKL contract and app/generated/ artifacts.