Python workflows

The Python SDK lets you turn application functions into observable workflows. Use it for ETL jobs, scheduled reports, application integrations, and reusable processing steps. You keep ordinary Python inside tasks and declare dependencies in a flow.

On this page

Define tasks and flows Choose how to execute Build reusable workflows Pass parameters safely Move to a workspace

Define tasks and flows

Python

from dagy import flow, task


@task
def normalize_email(value: str) -> str:
    return value.strip().lower()


@task
def make_contact(email: str) -> dict:
    return {"email": email, "source": "signup"}


@flow(name="prepare_contact")
def prepare_contact(email: str) -> dict:
    normalized = normalize_email(email)
    return make_contact(normalized)

make_contact depends on normalize_email because it consumes its output. Dependencies can be nested in lists, tuples, and dictionaries. Tasks without dependencies between them can run independently.

During graph construction, a task call returns an output handle rather than its computed value. Pass that handle to another task. Move indexing, arithmetic, filtering, and other operations on the eventual result inside a task. Do not use ordinary Python if to inspect an unresolved output; use conditional branches.

Choose distinct task names within a flow. By default, names are function names and descriptions come from docstrings. Define importable functions at module level and keep source imports free of deployment or execution side effects.

Choose how to execute

Call	Behavior	Result
`prepare_contact(email="person@example.com")`	Eager Python execution with decorator retries, hooks, and parameter checks	The flow function's return value
`prepare_contact.build(email="person@example.com")`	Builds the graph without executing tasks	`FlowSpec`
`prepare_contact.run_local(email="person@example.com")`	Executes the graph locally and records task results and logs	`LocalRunContext`
`prepare_contact.run(email="person@example.com")`	Triggers a hosted run if an API URL is configured; otherwise runs locally	Hosted response dictionary or local context
`prepare_contact.deploy(...)`	Builds and registers a deployment if an API URL is configured; otherwise runs locally	Deployment result or local context

Use explicit run_local() for local development. For hosted runs, prefer a deployment name through the CLI or the runs API. The SDK Flow.run() addresses the latest version by flow name, forwards keyword parameters, and does not forward positional arguments or set an environment for you.

A direct eager call does not enforce DAG scheduling, task timeouts, or worker concurrency and does not create tracked local run history. DAG-only branching requires build() or run_local() rather than an eager call. The decorators execute synchronous Python functions; the task runner does not await an async def task. Use a synchronous wrapper around an async client or the separate async node framework.

Build reusable workflows

Use task outputs to connect steps. Fan-out and fan-in let independent operations feed a shared result.
Use .map() for a collection discovered at runtime and .expand() for zipped inputs. Their output is an ordered list available to a downstream task.
Invoke a decorated child flow inside a parent flow to compose workflows. The child is built with its resolved input values at execution time.
Attach retries, timeouts, and hooks to tasks that call unreliable services.
Add quality expectations and data contracts before publishing results.
Declare input and output assets to describe which datasets each task reads and writes.

Pass parameters safely

Flow arguments become run parameters. Python calls retain their Python types; CLI --param key=value passes strings. Type validation checks function signatures and supported type annotations without coercion. Pass count=10 in Python rather than count="10" for an integer parameter.

Build-time literals support JSON primitives and collections, dates/times, timedeltas, bytes, Decimal, enum members, sets, tuples, and dataclasses. Custom enum and dataclass types must be importable in the execution environment. Arbitrary objects raise SerializationError when building. Literal dictionary keys $ref, $type, $v, and $repr are reserved by the SDK and cannot be used in captured arguments.

Prefer JSON-compatible values for data exchanged between tasks and external systems. The richer build-time literal support does not mean every Python object can be persisted as an output artifact. Keep credentials out of run parameters, source files, and printed output.

Move to a workspace

First connect and authenticate, then deploy the workflow. Set an explicit API URL and environment so a missing profile cannot silently turn a deploy into a local run.

The current build packages the flow source file and its graph definition. It records dependencies from your project configuration without installing or vendoring them. Imported local modules and data files are not automatically included. Use a self-contained first flow and dependency packages for deployed libraries.

For method signatures, defaults, result types, and restrictions, see the SDK reference.