Handlers

A Handler is the HTTP face of your app. When Atlan needs to test connection credentials, run preflight checks, or discover available metadata, it sends HTTP requests to your app's handler service. The Handler receives those requests, validates the incoming data with typed contracts, performs the work, and returns a structured response.

The division of responsibility is straightforward:

• The App handles durable execution—it runs tasks that extract, transform, or publish metadata, and automatically recovers from failures.
• The Handler handles the HTTP protocol—it validates credentials, checks connectivity, and surfaces metadata for the app configuration UI.

Both run as separate services in production, but share the same codebase and can run together in a single process during local development.

Implement handler

Every handler subclasses Handler from application_sdk.handler and implements three methods:

from application_sdk.handler import Handler
from application_sdk.handler.contracts import (
    AuthInput, AuthOutput, AuthStatus,
    PreflightInput, PreflightOutput, PreflightStatus,
    MetadataInput, MetadataOutput,
)

class MyHandler(Handler):
    async def test_auth(self, input: AuthInput) -> AuthOutput:
        ...

    async def preflight_check(self, input: PreflightInput) -> PreflightOutput:
        ...

    async def fetch_metadata(self, input: MetadataInput) -> MetadataOutput:
        ...

Each method receives a single typed Input and returns a single typed Output. The framework validates both at the boundary—malformed requests are rejected before your code runs.

The framework injects self.context before each method call and clears it after, making handlers stateless and safe for concurrent requests.

Methods

Each method receives a single typed Input and returns a typed Output. Raise a typed error from application_sdk.errors for protocol failures—the framework maps it to the appropriate HTTP status automatically. Pydantic ValidationError on the request body is automatically caught and returned as 400.

Error class	HTTP status	When to use
InvalidInputError	400	Missing or malformed credentials/config in the request body
AuthError	401	Credentials present but invalid; token expired
AppPermissionDeniedError	403	Auth succeeded but the role lacks required permissions
NotFoundError	404	Target database, schema, or resource doesn't exist
RateLimitedError	429	Request throttled by the target system
DependencyUnavailableError	503	Target is unreachable (network down, DNS failure)
AppTimeoutError	504	Request exceeded its timeout
InternalError	500	Genuine SDK or connector bug (last resort)

test_auth

Tests whether the provided credentials are valid.

class AuthInput:
    credentials: list[HandlerCredential] = []  # key/value credential pairs
    connection_id: str = ""                    # optional connection ID
    timeout_seconds: int = 30

class AuthOutput:
    status: AuthStatus        # SUCCESS, FAILED, EXPIRED, or INVALID_CREDENTIALS
    message: str = ""         # optional detail visible to the user
    identities: list[str] = []   # verified identities (usernames, roles)
    scopes: list[str] = []       # authorized scopes or permissions
    expires_at: str = ""         # ISO-8601 expiry timestamp

When credentials are present but authentication fails (wrong password, expired token, account locked), return AuthOutput with status=AuthStatus.FAILED rather than raising. This gives the UI a message it can display directly to the user:

async def test_auth(self, input: AuthInput) -> AuthOutput:
    try:
        async with connect(...) as conn:
            await conn.execute("SELECT 1")
        return AuthOutput(status=AuthStatus.SUCCESS)
    except AuthenticationError as exc:
        return AuthOutput(
            status=AuthStatus.FAILED,
            message=f"Connection failed: {exc}",
        )

preflight_check

Runs one or more named connectivity checks before an app runs.

class PreflightInput:
    credentials: list[HandlerCredential] = []
    connection_config: dict[str, Any] = {}  # host, port, database, etc.
    checks_to_run: list[str] = []           # empty list = run all checks
    timeout_seconds: int = 60

class PreflightOutput:
    status: PreflightStatus            # READY, NOT_READY, or PARTIAL
    checks: list[PreflightCheck] = []  # individual check results
    message: str = ""
    total_duration_ms: float = 0.0

class PreflightCheck:
    name: str      # machine-readable check identifier
    passed: bool
    message: str = ""

Raise typed errors to give callers semantic HTTP status codes they can act on without parsing message strings:

from application_sdk.errors import (
    AppPermissionDeniedError,
    DependencyUnavailableError,
    InvalidInputError,
)

class MyHandler(Handler):
    async def preflight_check(self, input: PreflightInput) -> PreflightOutput:
        if not self.context.has_credential("api_key"):
            raise InvalidInputError("api_key credential is required")

        try:
            async with connect(self.context.get_credential("host")) as conn:
                await conn.execute("SELECT 1")
        except ConnectionRefusedError as exc:
            raise DependencyUnavailableError("Warehouse is unreachable") from exc
        except PermissionDeniedError as exc:
            raise AppPermissionDeniedError("Role lacks SELECT on information_schema") from exc

        return PreflightOutput(status=PreflightStatus.READY)

When some checks pass and others fail, return PreflightStatus.PARTIAL so the UI can show the user exactly what's and isn't working:

async def preflight_check(self, input: PreflightInput) -> PreflightOutput:
    checks = [...]  # run each check independently
    passed = sum(1 for c in checks if c.passed)
    if passed == len(checks):
        status = PreflightStatus.READY
    elif passed == 0:
        status = PreflightStatus.NOT_READY
    else:
        status = PreflightStatus.PARTIAL
    return PreflightOutput(status=status, checks=checks)

fetch_metadata

Returns the set of schemas or objects available for selection in the Atlan app configuration UI. Two specialized output types are provided for the most common cases:

class MetadataInput:
    credentials: list[HandlerCredential] = []
    connection_config: dict[str, Any] = {}
    object_filter: str = ""     # filter pattern, e.g. "public.*"
    include_fields: bool = True
    max_objects: int = 1000
    timeout_seconds: int = 120

# For SQL connectors (populates the sqltree widget)
class SqlMetadataOutput(MetadataOutput):
    objects: list[SqlMetadataObject] = []

class SqlMetadataObject:
    TABLE_CATALOG: str = ""
    TABLE_SCHEMA: str = ""

# For API connectors (populates the apitree widget)
class ApiMetadataOutput(MetadataOutput):
    objects: list[ApiMetadataObject] = []

Context in handlers

self.context is injected per request and cleared after—it's stateless across requests. It provides access to the credentials submitted in the request body and to the framework's secret store. This is the same self.context interface available inside @task methods on an App.

Credentials

Credentials arrive as key/value pairs in the request body. Access them by key:

async def test_auth(self, input: AuthInput) -> AuthOutput:
    # Check whether a key is present before using it
    if not self.context.has_credential("api_key"):
        raise InvalidInputError("api_key is required")

    # Retrieve a specific credential value
    api_key = self.context.get_credential("api_key")

    # Access all credentials as a list
    all_creds = self.context.credentials

Other secrets

For secrets not tied to request credentials, use get_secret:

async def test_auth(self, input: AuthInput) -> AuthOutput:
    signing_key = await self.context.get_secret("my-signing-key")

Full example

The following handler validates credentials by attempting a real connection, runs two named preflight checks, and returns SQL schema objects for the metadata browser:

from application_sdk.handler import Handler
from application_sdk.handler.contracts import (
    AuthInput, AuthOutput, AuthStatus,
    PreflightInput, PreflightOutput, PreflightStatus, PreflightCheck,
    MetadataInput, SqlMetadataOutput, SqlMetadataObject,
)
from application_sdk.errors import InvalidInputError


class WarehouseHandler(Handler):
    async def test_auth(self, input: AuthInput) -> AuthOutput:
        host = self.context.get_credential("host")
        username = self.context.get_credential("username")
        password = self.context.get_credential("password")

        if not host or not username or not password:
            raise InvalidInputError("host, username, and password credentials are required")

        try:
            async with create_connection(host, username, password) as conn:
                await conn.execute("SELECT 1")
            return AuthOutput(status=AuthStatus.SUCCESS)
        except Exception as exc:
            return AuthOutput(
                status=AuthStatus.FAILED,
                message=f"Connection failed: {exc}",
            )

    async def preflight_check(self, input: PreflightInput) -> PreflightOutput:
        host = self.context.get_credential("host")
        username = self.context.get_credential("username")
        password = self.context.get_credential("password")

        schema_ok = False
        tables_ok = False
        try:
            async with create_connection(host, username, password) as conn:
                schemas = await conn.execute(
                    "SELECT COUNT(*) FROM information_schema.schemata"
                )
                schema_ok = True
                tables = await conn.execute(
                    "SELECT COUNT(*) FROM information_schema.tables"
                )
                tables_ok = True
        except Exception:
            pass

        checks = [
            PreflightCheck(
                name="schemaCheck",
                passed=schema_ok,
                message="Schema access check successful" if schema_ok else "Could not query schemas",
            ),
            PreflightCheck(
                name="tableCheck",
                passed=tables_ok,
                message="Table access check successful" if tables_ok else "Could not query tables",
            ),
        ]
        status = PreflightStatus.READY if all(c.passed for c in checks) else PreflightStatus.NOT_READY
        return PreflightOutput(status=status, checks=checks)

    async def fetch_metadata(self, input: MetadataInput) -> SqlMetadataOutput:
        host = self.context.get_credential("host")
        username = self.context.get_credential("username")
        password = self.context.get_credential("password")

        async with create_connection(host, username, password) as conn:
            rows = await conn.execute(
                "SELECT TABLE_CATALOG, TABLE_SCHEMA FROM information_schema.schemata"
            )
        return SqlMetadataOutput(
            objects=[
                SqlMetadataObject(
                    TABLE_CATALOG=row["TABLE_CATALOG"],
                    TABLE_SCHEMA=row["TABLE_SCHEMA"],
                )
                for row in rows
            ]
        )

See also

• Credentials: typed credential references and self.context.resolve_credential()
• Apps and tasks: how the App and its tasks work alongside the Handler
• Built-in APIs: full API reference for handler contracts and the built-in server layer