Skip to main content

Workflow tool with Python code steps and webhooks

In Windmill, each step of a flow can be a Python function, and every flow gets webhook URLs that start it with the request body as input. Steps import any PyPI package: Windmill reads the imports when you save and pins them in a lockfile, so there is no requirements file to maintain. The example below receives an order from a shop's webhook, validates it with pydantic in one Python step, and posts a summary to Slack with requests in a second step.

1. The flow​

The flow takes one input, order, and has two Python steps. This is its OpenFlow definition, as the CLI stores it in f/shop/order_to_slack.flow/flow.yaml; the flow editor builds the same thing.

summary: New order webhook to Slack
schema:
  type: object
  properties:
    order:
      type: object
  required: [order]
value:
  modules:
    - id: a
      summary: Validate the order
      value:
        type: rawscript
        language: python3
        content: |
          from pydantic import BaseModel


          class Item(BaseModel):
              sku: str
              quantity: int
              unit_price: float


          class Order(BaseModel):
              id: str
              customer_email: str
              items: list[Item]


          def main(order: dict):
              parsed = Order.model_validate(order)
              total = sum(i.quantity * i.unit_price for i in parsed.items)
              return {"id": parsed.id, "customer": parsed.customer_email, "total": round(total, 2)}
        input_transforms:
          order:
            type: javascript
            expr: flow_input.order
    - id: b
      summary: Post to Slack
      value:
        type: rawscript
        language: python3
        content: |
          import requests
          import wmill


          def main(summary: dict):
              url = wmill.get_variable("f/shop/slack_webhook_url")
              text = f"New order {summary['id']} from {summary['customer']}: {summary['total']} EUR"
              r = requests.post(url, json={"text": text}, timeout=10)
              r.raise_for_status()
              return {"posted": text, "status": r.status_code}
        input_transforms:
          summary:
            type: javascript
            expr: results.a
  • Step a gets the webhook's order field through flow_input.order, and step b gets step a's return value through results.a.
  • pydantic and requests are not declared anywhere else: Windmill resolves the imports into a lockfile per step.
  • The Slack webhook URL is a secret variable at f/shop/slack_webhook_url, read with wmill.get_variable, not a flow input.

Generate the lockfiles and deploy from the CLI:

wmill generate-metadata f/shop/order_to_slack.flow --yes
wmill flow push f/shop/order_to_slack.flow f/shop/order_to_slack

2. Call the webhook​

Every flow has an asynchronous and a synchronous webhook. The synchronous one waits for the flow and returns its result:

curl -X POST "https://<your-instance>/api/w/<workspace>/jobs/run_wait_result/f/f/shop/order_to_slack" \
  -H "Authorization: Bearer $WINDMILL_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{"order": {"id": "A-1042", "customer_email": "[email protected]",
       "items": [{"sku": "MUG", "quantity": 2, "unit_price": 12.5},
                 {"sku": "TEE", "quantity": 1, "unit_price": 20}]}}'
{"posted": "New order A-1042 from [email protected]: 45.0 EUR", "status": 200}

The top-level keys of the JSON body become the flow's inputs, here order. If the order is missing customer_email, step a raises pydantic's validation error and the response carries the error, with a message that starts 1 validation error for Order, names the missing field, and includes "step_id": "a".

The asynchronous endpoint, /api/w/<workspace>/jobs/run/f/f/shop/order_to_slack, returns a job id at once; use it when the sender should not wait, which is what most webhook senders expect.

For the caller's token, create a webhook-specific token, which can run this flow and nothing else. If the sender cannot add an Authorization header or needs its own URL path, put an HTTP route in front of the flow, with its own authentication options and a preprocessor to reshape the request.

3. Run, debug and retry​

  • Each run is in the runs page with the request body, each step's logs and result, and the error if one failed.
  • You can test a step on its own in the flow editor with sample input, without calling the webhook.
  • Add retries to step b so a Slack outage does not lose the order, and a flow error handler to be told when a run fails.
  • The same flow can also run on a schedule, from a form, or from other triggers.

Steps do not have to be Python: a flow can mix Python, TypeScript, Bash, SQL and the other supported languages.

Edition notes​

Flows, Python steps, webhooks, HTTP routes, variables and step retries are in the open-source Community Edition. On a self-hosted Community Edition, the runs history is kept for up to 30 days; Kafka, NATS, SQS and GCP Pub/Sub triggers, concurrency limits and external secret backends need the Enterprise Edition.

Webhooks
Sync and async endpoints, tokens and custom response codes.
Python quickstart
Write Python scripts with dependencies resolved from imports.