Build your first app

In the previous tutorial you ran the Hello World app and read how it works. Now you evolve it—step by step, adding a new input field via the Pkl contract, threading it through a task, chaining in a third task, and pinning everything with a round-trip test.

🎓

What you learn here: How the Pkl contract drives the input schema, how to chain a new task into an existing workflow, and how to keep contracts safe with serialization tests. By the end, you will have extended a working app end-to-end and verified it with tests.

Before you begin

Complete Run your first sample app before starting this tutorial. You need the pkl CLI installed—covered in Set up your development environment.

What you'll build

📦

New input field

Add a greeting_prefix field to contract/app.pkl — the single source of truth for the workflow's input schema.

⚡

Use the field in a task

Thread greeting_prefix through the generate_greetings task so each record uses the custom prefix instead of the hardcoded "Hello".

🔗

Third chained task

Add a format_summary task that takes the result of summarize and produces a formatted report string.

✅

Contract tests

Add round-trip serialization tests for the new contracts and run make test to verify everything works.

Add new input field via Pkl

The top-level workflow input is generated from contract/app.pkl. Open that file and add a greeting_prefix field inside the Greeting task block's inputs, after the repeat_count input definition:

["greeting_prefix"] = new TextInput {
  title = "Greeting prefix"
  helpText = "Word that precedes the name in each greeting (e.g. Hello, Hi, Hey)."
  required = false
  default = "Hello"
  placeholderText = "Hello"
}

Save the file, then regenerate the Python code and UI manifest:

make generate

Open app/generated/_input.py. You'll see a new field on AppInputContract:

greeting_prefix: str = "Hello"
"""Word that precedes the name in each greeting (e.g. Hello, Hi, Hey)."""

Because HelloWorldInput = AppInputContract in app/contracts.py, the field is now available on every HelloWorldInput instance—no further changes to contracts.py needed for the top-level input.

What make generate does: It runs pkl eval on contract/app.pkl, which rewrites app/generated/_input.py and app/generated/hello-world.json. The JSON file is what the Atlan UI reads to build the run form. Both files are auto-generated — never hand-edit them.

Use your new field

The generate_greetings task builds the greeting records. You need to:

Add greeting_prefix to GenerateGreetingsInput in app/contracts.py
Use it in the task body in app/connector.py
Thread it through from run()

In app/contracts.py, add the field to GenerateGreetingsInput:

class GenerateGreetingsInput(Input):
    name: str = "World"
    repeat_count: int = 1
    output_dir: str = ""
    greeting_prefix: str = ""   # ← add this

In app/connector.py, update the record generation inside generate_greetings to use the prefix:

record = {"index": i, "message": f"{input.greeting_prefix}, {input.name}!"}

In run() inside HelloWorldApp, pass the new field when constructing GenerateGreetingsInput:

greetings = await self.generate_greetings(
    GenerateGreetingsInput(
        name=input.name,
        repeat_count=input.repeat_count,
        greeting_prefix=input.greeting_prefix,   # ← add this
        output_dir=output_dir,
    )
)

Start the app, then open a second terminal and send a request with greeting_prefix set:

make run

In the second terminal:

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

Fetch the result using the workflow_id from the response:

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

The greeting records use "Hi" as the prefix.

Add your third task

The current chain is generate_greetings → summarize. Add a third task, format_summary, that takes the summary and produces a formatted one-line report string.

In app/contracts.py, add new Input and Output classes for the new task:

class FormatSummaryInput(Input):
    message: str = ""
    record_count: int = 0
    greeting_prefix: str = ""


class FormatSummaryOutput(Output):
    report: str = ""

Also add report: str = "" to HelloWorldOutput so it appears in the workflow result:

class HelloWorldOutput(Output):
    message: str = ""
    record_count: int = 0
    output_file: FileReference | None = None
    report: str = ""   # ← add this

In app/connector.py, add the import at the top of the file and the new task method to HelloWorldApp:

from app.contracts import (
    FormatSummaryInput,    # ← add to existing import
    FormatSummaryOutput,   # ← add to existing import
    ...
)

@task
async def format_summary(self, input: FormatSummaryInput) -> FormatSummaryOutput:
    """Produce a one-line report string from the summarize output."""
    report = (
        f"Run complete — {input.greeting_prefix}, {input.message.split(', ')[-1]} "
        f"({input.record_count} record{'s' if input.record_count != 1 else ''} generated)"
    )
    self.logger.info("format_summary completed report=%s", report)
    return FormatSummaryOutput(report=report)

In run(), chain format_summary after summarize and include report in the return value:

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,
            greeting_prefix=input.greeting_prefix,
            output_dir=output_dir,
        )
    )

    summary = await self.summarize(
        SummarizeInput(greetings_file=greetings.greetings_file)
    )

    fmt = await self.format_summary(
        FormatSummaryInput(
            message=summary.message,
            record_count=summary.record_count,
            greeting_prefix=input.greeting_prefix,
        )
    )

    return HelloWorldOutput(
        message=summary.message,
        record_count=summary.record_count,
        output_file=greetings.greetings_file,
        report=fmt.report,
    )

Start the app again, then trigger a run from a second terminal:

make run

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

Look for log lines from generate_greetings, summarize, and format_summary completing in sequence.

Update unit tests

Any contract that crosses a task boundary must have a round-trip serialization test. Open tests/unit/test_contracts.py and add a test class for the new contracts at the bottom:

from app.contracts import (
    ...,
    FormatSummaryInput,
    FormatSummaryOutput,
)

class TestFormatSummaryContracts:
    def test_input_defaults(self) -> None:
        decoded = _round_trip(FormatSummaryInput(), FormatSummaryInput)
        assert decoded.message == ""
        assert decoded.record_count == 0
        assert decoded.greeting_prefix == ""

    def test_input_values(self) -> None:
        original = FormatSummaryInput(message="World!", record_count=5, greeting_prefix="Hi")
        decoded = _round_trip(original, FormatSummaryInput)
        assert decoded.message == "World!"
        assert decoded.record_count == 5
        assert decoded.greeting_prefix == "Hi"

    def test_output_round_trip(self) -> None:
        original = FormatSummaryOutput(report="Run complete — Hi, World! (5 records generated)")
        decoded = _round_trip(original, FormatSummaryOutput)
        assert decoded.report == "Run complete — Hi, World! (5 records generated)"

Also update the existing TestHelloWorldOutput class to cover the new report field:

def test_with_values_round_trip(self) -> None:
    original = HelloWorldOutput(
        message="Hello, Atlan!",
        record_count=3,
        output_file=_sample_file_ref("/tmp/out.jsonl"),
        report="Run complete — Hello, Atlan! (3 records generated)",
    )
    decoded = _round_trip(original, HelloWorldOutput)
    assert decoded.report == "Run complete — Hello, Atlan! (3 records generated)"

Also update the existing connector tests

Adding greeting_prefix to GenerateGreetingsInput changes the message format in generate_greetings, which breaks two existing tests in tests/unit/test_connector.py. Update both calls to pass greeting_prefix explicitly:

# test_writes_one_record_by_default
out = await hello_app.generate_greetings(
    GenerateGreetingsInput(name="World", repeat_count=1, output_dir=str(tmp_path), greeting_prefix="Hello"),
)

# test_repeat_count_writes_n_records
out = await hello_app.generate_greetings(
    GenerateGreetingsInput(name="Atlan", repeat_count=4, output_dir=str(tmp_path), greeting_prefix="Hello"),
)

Run the test suite:

make test

All tests pass. If any fail, the error message points to the contract that didn't survive the round trip.

What you built

🎉

You evolved a working app end-to-end. You added a field to the Pkl contract, regenerated the Python schema, threaded the field through a task, chained in a third task, pinned the new contracts with tests, and verified the result over HTTP.

Here is what each step demonstrated:

Pkl contract as single source of truth: contract/app.pkl drives both the Atlan UI form and the Python typed input. Changing the contract in one place regenerates both.
Adding a field to a task contract: GenerateGreetingsInput is a hand-written contract separate from the top-level input. These app-internal contracts aren't exposed outside the app, so you are free to make breaking changes to them—they evolve independently of the public input schema.
Chaining a third task: run() is just Python await calls; adding a step is adding one method and one await.
Round-trip tests as a safety net: catching schema-breaking changes before they reach CI or production.

What's next

Now that you have built and extended your first complete app, explore the concept pages to go deeper:

Apps and tasks: Multiple entry points with @entrypoint, parallel tasks, lifecycle hooks, and determinism rules.
Handlers: The HTTP surface (test_auth, preflight_check, fetch_metadata) that Atlan calls before triggering a run.
Configuration: Production CLI modes (handler, worker, combined) and environment variables.

Before you begin​

What you'll build​

New input field

Use the field in a task

Third chained task

Contract tests

Add new input field via Pkl​

Use your new field​

Add your third task​

Update unit tests​

What you built​

What's next​