
## Build your first app

URL: https://docs.atlan.com/product/capabilities/build-apps/tutorials/build-first-app

> Edit the Pkl contract to add a new input field, use it in a task, add a third task, and pin the contracts with tests.

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](./run-sample-app.md) before starting this tutorial. You need the `pkl` CLI installed—covered in [Set up your development environment](./set-up-development-environment.md).

## What you'll build

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

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

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

 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:

```pkl
["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:

```bash
make generate
```

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

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

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

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

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

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

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

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

```bash
make run
```

In the second terminal:

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

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

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

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

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

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

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

```bash
make run
```

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

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

```python
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 — see full content on the documentation site.*

Run the test suite:

```bash
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](../concepts/apps-and-tasks.md)**: Multiple entry points with `@entrypoint`, parallel tasks, lifecycle hooks, and determinism rules.
- **[Handlers](../concepts/handlers.md)**: The HTTP surface (`test_auth`, `preflight_check`, `fetch_metadata`) that Atlan calls before triggering a run.
- **[Configuration](../concepts/configuration.md)**: Production CLI modes (`handler`, `worker`, `combined`) and environment variables.

---
