Skip to main content

Run your first sample app

Think of this tutorial like touring a model home before building your own. You clone a complete, working application, get it running in two commands, and read through the code to see exactly how it's structured.

🎓

What you learn here: How to clone the Hello World app, run it locally, and understand its core building blocks — App, @task, typed contracts, and the Pkl-driven input schema.

Before you begin​

Complete Set up your development environment before starting this tutorial.

Core concepts​

Before running any commands, here are the building blocks you'll see in action:

🏗️

App

The blueprint. run() sequences your tasks and defines how results flow between them. Must stay deterministic — all side effects go in tasks.

⚡

@task

Where side effects live — API calls, file I/O, data processing. Results are checkpointed so failures resume from the last completed step.

📋

Contracts

Typed Input/Output boundaries at every task boundary. Validated by the framework and used by the Atlan UI to build the run form.

Clone your sample app​

git clone https://github.com/atlanhq/atlan-hello-world-app
cd atlan-hello-world-app
Explore the project structure
atlan-hello-world-app/
├── Makefile                   # generate / test / run / lint shortcuts
├── pyproject.toml             # Python project — pins atlan-application-sdk
├── Dockerfile                 # production container
├── atlan.yaml                 # app manifest (deploy config)
├── app.yaml                   # image binding (CI fills in the tag)
├── .env.example               # documented dev env vars
├── main.py                    # container entry point (use make run for dev)
├── contract/
│   ├── PklProject             # Pkl package dependencies (pins app-contract-toolkit version)
│   ├── PklProject.deps.json   # locked checksums for reproducible builds
│   └── app.pkl                # canonical input schema — single source of truth
├── app/
│   ├── connector.py           # HelloWorldApp(App) + @task methods
│   ├── contracts.py           # typed Input/Output for every task
│   ├── run_dev.py             # local dev server entry point
│   └── generated/             # AUTO-GENERATED from contract/app.pkl
│       ├── _input.py          # AppInputContract — the top-level workflow input
│       ├── hello-world.json   # UI form definition
│       └── manifest.json      # app manifest for the Atlan platform
└── tests/
    └── unit/
        ├── test_contracts.py  # round-trip tests for every contract
        └── test_connector.py  # task logic tested with a fake context

The connector logic lives entirely in app/connector.py. app/contracts.py defines the typed inputs and outputs for each task. The top-level input (what the Atlan UI form and POST /workflows/v1/start accept) is autogenerated from contract/app.pkl.

Set up and run​

Install dependencies​

Install the project's Python dependencies (and the Python toolchain itself if needed):

uv sync

Start dev server​

make run

On the first run, the SDK downloads and starts its runtime components (Dapr and an embedded Temporal engine) and caches them locally. Subsequent runs are instant.

When startup is complete, you can see a Dev server running at http://127.0.0.1:8000 banner listing the available endpoints, followed by:

Combined mode started: app=hello-world queue=hello-world-app-queue port=8000

The server is now ready to accept requests. The example_input defined in app/run_dev.py is printed as the example curl command in the banner—it doesn't trigger a run automatically. You send the first request in the next section.

Send request​

Open a second terminal and send a request:

curl -X POST http://127.0.0.1:8000/workflows/v1/start \
  -H "Content-Type: application/json" \
  -d '{"name": "Atlan", "repeat_count": 3}'

The response contains a workflow_id. Use it to fetch the result once the run finishes:

curl http://127.0.0.1:8000/workflows/v1/result/<workflow_id>

You'll see a completed status with a message of "Hello, Atlan!" and a record_count of 3.

Deep dive​

Each run calls HelloWorldApp.run() end-to-end:

  1. run() received the input (name, repeat_count) as a typed HelloWorldInput contract.
  2. generate_greetings wrote repeat_count greeting records to a JSONL file and returned a FileReference.
  3. summarize read that file and returned a message and count.

run() itself did no I/O—it only composed the task calls and routed data between them.

App and run()​

Every Atlan app is a class that extends App. The run() method is its heart: it defines the app's input contract (what fields the caller must provide) and its output contract (what the result looks like), and it sequences the tasks that do the actual work. Think of it as pure orchestration: run() decides what happens and in what order, while @task methods decide how.

The one rule that follows from this is that run() must stay deterministic. The SDK can replay it on retry, so it must always produce the same sequence of task calls given the same input. That means no API calls, no file I/O, no datetime.now() directly in run(): all side effects belong in tasks.

See the code
class HelloWorldApp(App):
    name = "hello-world"

    async def run(self, input: HelloWorldInput) -> HelloWorldOutput:
        output_dir = str(Path(tempfile.gettempdir()) / "hello-world" / self.run_id)
        greetings = await self.generate_greetings(
            GenerateGreetingsInput(
                name=input.name,
                repeat_count=input.repeat_count,
                output_dir=output_dir,
            )
        )
        summary = await self.summarize(
            SummarizeInput(greetings_file=greetings.greetings_file)
        )
        return HelloWorldOutput(
            message=summary.message,
            record_count=summary.record_count,
            output_file=greetings.greetings_file,
        )

@task​

@task is what makes a method durable. When a task completes, the SDK checkpoints its result—so if the process restarts mid-run, completed tasks are skipped and execution picks up from the last checkpoint. Unlike run(), task methods have no determinism requirement; they can do anything a normal Python async function can do.

The three timeout parameters on each task are worth understanding:

  • timeout_seconds: the maximum time the task may run in total before the SDK considers it timed out.
  • heartbeat_timeout_seconds: how long it can go without sending a heartbeat before the SDK considers it stuck.
  • auto_heartbeat_seconds: tells the SDK to send heartbeats automatically on the task's behalf, so long-running tasks don't need to heartbeat manually.
See the code
@task(
    timeout_seconds=60,
    heartbeat_timeout_seconds=30,
    auto_heartbeat_seconds=10,
)
async def generate_greetings(
    self, input: GenerateGreetingsInput
) -> GenerateGreetingsOutput:
    # Write repeat_count greeting records to a JSONL file.
    ...
    return GenerateGreetingsOutput(
        greetings_file=FileReference(local_path=str(out_path), tier=StorageTier.RETAINED),
        record_count=input.repeat_count,
    )

Contracts and persistence​

Every @task takes exactly one typed Input and returns exactly one typed Output: both Pydantic-based classes from the SDK. This isn't just convention: the SDK serializes and checkpoints these objects at every task boundary, which is how replay and schema evolution stay safe.

FileReference is the SDK's typed pointer to a file a task produced. Rather than passing a raw path string between tasks, you pass a FileReference: which lets the SDK manage the file's lifecycle (local retention in dev, remote object storage in production) without any changes to your task logic.

See the code
class GenerateGreetingsInput(Input):
    name: str = "World"
    repeat_count: int = 1
    output_dir: str = ""

class GenerateGreetingsOutput(Output):
    greetings_file: FileReference | None = None
    record_count: int = 0

class SummarizeInput(Input):
    greetings_file: FileReference | None = None

class SummarizeOutput(Output):
    message: str = ""
    record_count: int = 0

Input contract​

When a user runs your app from the Atlan UI, they fill in a form. When a system triggers it programmatically, it sends a JSON body. Both surfaces must agree on exactly the same fields—and keeping two hand-written definitions in sync is a maintenance problem waiting to happen. So instead, both are generated from a single definition: contract/app.pkl.

contract/app.pkl declares what fields exist, their types, their defaults, and the labels the UI must display. Running make generate turns that into both a typed Python dataclass (app/generated/_input.py) and a UI form definition (app/generated/hello-world.json)—so the form and the code are guaranteed to stay in sync. The @app-contract-toolkit import alias is resolved via contract/PklProject, which pins the toolkit version. The next tutorial walks through adding a field to the contract.

See the code
// contract/PklProject — pins the app-contract-toolkit version
amends "pkl:Project"
dependencies {
  ["app-contract-toolkit"] {
    uri = "package://atlanhq.github.io/application-sdk/contracts/[email protected]"
  }
}
// contract/app.pkl
amends "@app-contract-toolkit/App.pkl"

name = "hello-world"
displayName = "Hello World"
hasCredentialConfig = false

uiConfig = new UIConfig {
  tasks {
    ["Greeting"] {
      description = "Configure the greeting this workflow produces."
      inputs {
        ["name"] = new TextInput {
          title = "Name"
          helpText = "Who should the workflow greet?"
          required = true
          default = "World"
          placeholderText = "World"
        }
        ["repeat_count"] = new NumericInput {
          title = "Repeat count"
          helpText = "How many greeting records to generate."
          required = false
          default = 1
        }
      }
    }
  }
}

run_dev_combined()​

In production, the HTTP handler and the workflow worker run as separate processes. run_dev_combined() collapses them into one for local development—along with an in-process workflow runtime—so you can iterate without managing multiple processes or external services. The example_input you pass in populates the example curl command printed in the startup banner, but doesn't trigger a run automatically; you send the first request yourself.

See the code
await run_dev_combined(
    HelloWorldApp,
    example_input={
        "name": "World",
        "repeat_count": 1,
    },
)

Observe retries (optional)​

To see the SDK's automatic retry behavior, temporarily add a failure to generate_greetings in app/connector.py:

async def generate_greetings(self, input: GenerateGreetingsInput) -> GenerateGreetingsOutput:
    raise RuntimeError("Simulated failure")

Restart the dev server to pick up the change (Ctrl+C, then make run again), then in a second terminal trigger a new run:

curl -X POST http://127.0.0.1:8000/workflows/v1/start \
  -H "Content-Type: application/json" \
  -d '{"name": "Atlan", "repeat_count": 1}'

Log lines appear in the first terminal showing the task failing and the SDK retrying it automatically. If you fetch the result using the workflow_id from the response, you eventually see the run marked as failed once all retries are exhausted.

Revert the change before moving to the next tutorial.

🎉

You did it! You cloned a real App Framework application, ran it in two commands, and read through its core building blocks: App, chained @task methods, typed contracts, FileReference, and a Pkl-driven input schema.

What's next​

Next tutorial: Build your first app: edit the Pkl contract to add a new input field, use it in a task, add a third task to the chain, and pin the contract with a round-trip test.