AWS Workflow Python

Tip

This page is best viewed through the site search or through the AI.

Uses the Workflow component to create an AWS Lambda durable workflow using the Python runtime.

Hit the Invoker URL to start the workflow. The workflow logs a callback URL with a token query parameter. Open that URL to resume the waiting step.

sst.config.ts

const workflow = new sst.aws.Workflow("Workflow", {
  handler: "workflow/main.handler",
  runtime: "python3.13",
});

const resolver = new sst.aws.Function("Resolver", {
  handler: "resolver/main.handler",
  runtime: "python3.13",
  url: true,
  link: [workflow],
});

const invoker = new sst.aws.Function("Invoker", {
  handler: "invoker/main.handler",
  runtime: "python3.13",
  url: true,
  link: [workflow, resolver],
});

workflow/main.py

from typing import Any, Dict
from urllib.parse import urlencode

from aws_durable_execution_sdk_python import (
    DurableContext,
    StepContext,
    durable_execution,
    durable_step,
)
from aws_durable_execution_sdk_python.config import Duration, WaitForCallbackConfig


@durable_step
def start(step_context: StepContext) -> None:
    step_context.logger.info("Workflow started.")


@durable_step
def finish(step_context: StepContext, result: Any) -> None:
    step_context.logger.info(result)


@durable_execution
def handler(event: Dict[str, Any], context: DurableContext) -> None:
    context.step(start())

    def log_callback_url(token: str, step_context: StepContext) -> None:
        callback_url = f"{event['resolverUrl']}?{urlencode({'token': token})}"
        step_context.logger.info(callback_url)

    result = context.wait_for_callback(
        log_callback_url,
        name="callback",
        config=WaitForCallbackConfig(timeout=Duration.from_minutes(5)),
    )

    context.step(finish(result))

resolver/main.py

import boto3
import json
from typing import Any, Dict


lambda_client = boto3.client("lambda")


def handler(event: Dict[str, Any], _context: Any) -> str:
    query_params = event.get("queryStringParameters") or {}
    token = query_params.get("token")

    if not token:
        return "Missing token in query parameters"

    lambda_client.send_durable_execution_callback_success(
        CallbackId=token, Result=json.dumps({"message": "Sent from the resolver."})
    )

    return "Workflow callback sent. Check the logs to see the workflow succeed."

invoker/main.py

import boto3
import json
from typing import Any, Dict

from sst import Resource

client = boto3.client("lambda")


def handler(_event: Dict[str, Any], _context: Any) -> str:
    client.invoke(
        FunctionName=Resource.Workflow.name,
        Qualifier=Resource.Workflow.qualifier,
        InvocationType="Event",
        Payload=json.dumps({"resolverUrl": Resource.Resolver.url}),
    )

    return "Workflow started. Check the logs for the callback URL."

View the full example.