Quickstart: Python

This quickstart runs a workflow on your own machine: you create a pull app, run a worker that polls EZGH for work, and start a run from a script. Nothing needs a public endpoint. It takes about five minutes.

Before you start

• Python 3.12 or 3.13.
• The ezgh CLI, version 0.16.0 or later (install it, or run ezgh update), logged in.
• An organization ID and a project ID. Run ezgh whoami for the organization and ezgh projects list for the project, or copy them from the console. The SDKs take IDs, not slugs.
• An API key whose owner has WorkflowsFullAccess on the project, or the root user’s. Create one with ezgh iam api-keys create workflows-quickstart.

Set them in your shell:

export EZGH_API_KEY=ezgh_…
export EZGH_ORG_ID=<org-id>
export EZGH_PROJECT_ID=<project-id>

1. Create a pull app

ezgh workflows apps create quickstart --kind pull --project "$EZGH_PROJECT_ID"

The output shows the app. Keep its ID:

export EZGH_APP_ID=<ID>

2. Install the SDK

The SDK isn’t published to PyPI yet. Install it from GitHub by tag, in a virtual environment:

mkdir workflows-quickstart && cd workflows-quickstart
python3 -m venv .venv && . .venv/bin/activate
pip install git+https://github.com/ezgamehost/workflows-sdk-python@v0.1.0

It imports as ezgh_workflows, and has no dependencies outside the standard library.

3. Write a workflow

hello.py defines a workflow named hello with three steps: it computes a greeting, sleeps for five seconds, and records the time.

from ezgh_workflows import Context, workflow


@workflow("hello")
async def hello(ctx: Context) -> dict:
    greeting = await ctx.step.run("greet", lambda: f"Hello, {ctx.input['name']}!")
    await ctx.step.sleep("pause", "5s")
    at = await ctx.step.now("finished-at")
    return {"greeting": greeting, "at": at}

4. Run a worker

worker.py polls the app’s task queue and runs the workflows you give it. Each poll also tells EZGH which workflows this worker serves. Ctrl+C stops it.

import os

from ezgh_workflows import Worker

from hello import hello

worker = Worker(
    api_key=os.environ["EZGH_API_KEY"],
    organization_id=os.environ["EZGH_ORG_ID"],
    project_id=os.environ["EZGH_PROJECT_ID"],
    app=os.environ["EZGH_APP_ID"],
    workflows=[hello],
)
print("Worker polling. Press Ctrl+C to stop.")
worker.run_sync()

python worker.py

Leave it running.

5. Start a run

In a second terminal with the same environment variables and the virtual environment active, start.py starts a run, checks it every second until it ends, and prints its output:

import os
import time

from ezgh_workflows import Client

client = Client(
    api_key=os.environ["EZGH_API_KEY"],
    organization_id=os.environ["EZGH_ORG_ID"],
    project_id=os.environ["EZGH_PROJECT_ID"],
)
run = client.start(os.environ["EZGH_APP_ID"], "hello", {"name": "Ada"})
print("Started run", run["id"])
while run["status"] == "running":
    time.sleep(1)
    run = client.get(run["id"])
print(run["status"], client.output(run["id"]) if run["status"] == "completed" else run["error"])

python start.py

After about five seconds:

Started run fd47f6f0-91ea-80a5-b374-6a346a10af47
completed {'greeting': 'Hello, Ada!', 'at': '2026-10-11T20:30:52.860Z'}

The worker ran the workflow twice: once up to the sleep, and again after it, when the greet step returned its recorded result without running again.

A pull app starts runs only of workflows that one of its workers announced in the last 10 minutes. Starting a run with no worker running answers 409 workflow_not_announced.

6. Clean up

Stop the worker with Ctrl+C, then delete the app and, if you made one for this, the API key:

ezgh workflows apps delete quickstart --project "$EZGH_PROJECT_ID"
unset EZGH_API_KEY
ezgh iam api-keys delete workflows-quickstart

Next

• Concepts: what replay means for your code.
• Push and pull apps: serve workflows from an HTTPS endpoint instead.
• The SDK’s README lists its whole API.