## Handlers

URL: https://docs.atlan.com/product/capabilities/build-apps/concepts/handlers

> Implement the Handler—your app's HTTP interface—to validate credentials, run preflight checks, and surface available metadata for the Atlan setup form.

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:

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

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

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

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

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

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

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

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

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

```python
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](https://docs.atlan.com/llms/platform/build-apps/credentials/llms.txt): typed credential references and `self.context.resolve_credential()`
- [Apps and tasks](https://docs.atlan.com/llms/platform/build-apps/apps-and-tasks/llms.txt): how the App and its tasks work alongside the Handler
- [Built-in APIs](https://docs.atlan.com/llms/platform/build-apps/api/llms.txt): full API reference for handler contracts and the built-in server layer

---