Skip to main content

​
Quickstart

This guide will walk you through building a simple “Hello World” application using the Atlan Application SDK. You’ll learn the core concepts of workflows, activities, and the application lifecycle.

​
Prerequisites

Before you begin, make sure you have:
  • Python 3.11 or higher installed
  • The Atlan Application SDK installed (see Installation)

​
Build your first application

1

Install the SDK

Install the SDK with workflow support:
pip install atlan-application-sdk[workflows]
2

Create your application file

Create a new file called hello_world.py with the following code:
import asyncio
from datetime import timedelta
from typing import Any, Callable, Dict, Sequence, cast

from temporalio import activity, workflow

from application_sdk.activities import ActivitiesInterface
from application_sdk.activities.common.utils import auto_heartbeater
from application_sdk.application import BaseApplication
from application_sdk.observability.logger_adaptor import get_logger
from application_sdk.workflows import WorkflowInterface

APPLICATION_NAME = "hello-world"

logger = get_logger(__name__)


@workflow.defn
class HelloWorldWorkflow(WorkflowInterface):
    @workflow.run
    async def run(self, workflow_config: Dict[str, Any]) -> None:
        activities = HelloWorldActivities()

        await workflow.execute_activity_method(
            activities.demo_activity,
            args=[workflow_config],
            start_to_close_timeout=timedelta(seconds=10),
            heartbeat_timeout=timedelta(seconds=10),
        )

    @staticmethod
    def get_activities(activities: ActivitiesInterface) -> Sequence[Callable[..., Any]]:
        activities = cast(HelloWorldActivities, activities)
        return [
            activities.demo_activity,
        ]


class HelloWorldActivities(ActivitiesInterface):
    @activity.defn
    @auto_heartbeater
    async def demo_activity(self, workflow_args: Dict[str, Any]) -> Dict[str, Any]:
        return {"message": "Demo activity completed successfully"}


async def application_hello_world(daemon: bool = True) -> Dict[str, Any]:
    logger.info("Starting application_hello_world")

    # Initialize application
    app = BaseApplication(name=APPLICATION_NAME)

    # Setup workflow
    await app.setup_workflow(
        workflow_and_activities_classes=[(HelloWorldWorkflow, HelloWorldActivities)]
    )

    # Start workflow
    workflow_response = await app.start_workflow(
        workflow_args={"workflow_id": "hello-world"}, 
        workflow_class=HelloWorldWorkflow
    )

    # Start worker
    await app.start_worker(daemon=daemon)

    return workflow_response


if __name__ == "__main__":
    asyncio.run(application_hello_world(daemon=False))
3

Run the application

Run your application:
python hello_world.py
You should see log output indicating the workflow started and completed successfully:
INFO Starting application_hello_world
INFO Starting workflow execution
INFO Demo activity completed successfully

​
Understanding the code

Let’s break down the key components:

​
Workflow

The HelloWorldWorkflow class defines your workflow logic: 
@workflow.defn
class HelloWorldWorkflow(WorkflowInterface):
    @workflow.run
    async def run(self, workflow_config: Dict[str, Any]) -> None:
        # Workflow execution logic
  • @workflow.defn: Marks this class as a Temporal workflow
  • @workflow.run: Defines the main workflow execution method
  • Workflows orchestrate activities and contain business logic

​
Activities

The HelloWorldActivities class defines tasks that can be executed: 
class HelloWorldActivities(ActivitiesInterface):
    @activity.defn
    @auto_heartbeater
    async def demo_activity(self, workflow_args: Dict[str, Any]) -> Dict[str, Any]:
        return {"message": "Demo activity completed successfully"}
  • @activity.defn: Marks this method as a Temporal activity
  • @auto_heartbeater: Automatically sends heartbeats to Temporal during execution
  • Activities contain the actual work (API calls, database queries, etc.)

​
Application

The BaseApplication class manages the workflow lifecycle: 
app = BaseApplication(name=APPLICATION_NAME)

await app.setup_workflow(
    workflow_and_activities_classes=[(HelloWorldWorkflow, HelloWorldActivities)]
)

workflow_response = await app.start_workflow(
    workflow_args={"workflow_id": "hello-world"}, 
    workflow_class=HelloWorldWorkflow
)

await app.start_worker(daemon=daemon)
  • setup_workflow(): Registers your workflow and activities with Temporal
  • start_workflow(): Initiates a workflow execution
  • start_worker(): Starts the worker that processes workflow tasks

​
Next steps

Core concepts

Learn about the BaseApplication, workflows, and activities in depth

SQL application

Build a real-world application that extracts database metadata

Architecture

Understand how Temporal and Dapr power the SDK

API reference

Explore the complete API documentation

​
Troubleshooting

Make sure you installed the SDK with the workflows extra:
pip install atlan-application-sdk[workflows]
The SDK requires a Temporal server to be running. For local development, you can:
  1. Use the provided Docker Compose setup in the repository
  2. Install Temporal CLI and run temporal server start-dev
The SDK requires Python 3.11 or higher. Check your version:
python --version

Build docs developers (and LLMs) love

Get started for freeTalk to us