Cloud functions as pipeline steps

An HTTPS endpoint can call anything, but the most useful thing to point it at is a small function you wrote — in GCP Cloud Functions, AWS Lambda or Azure Functions — doing the one thing your pipeline needs that SQL cannot.

A function written without knowing what the caller expects usually works on the first manual test and then disappoints in a pipeline. This page is the contract.

Four rules
Return JSON A flat object is the useful shape — see below
Answer within 30 seconds The default timeout. 120 seconds is the maximum you can set
Use HTTPS Plain http:// is refused when the endpoint is saved
Do not redirect A 3xx is never followed. The step fails and tells you where it pointed

What to return

The response body becomes pipeline data when Keep the response as pipeline data is ticked, and the next step can read its fields.

Return a flat JSON object. It becomes one row, and each key becomes a field a later step can bind to or branch on:

{ "job_id": "7c2f", "status": "running", "rows_written": 0 }

A JSON array of objects becomes one row each, which is right when the function returns a list. Anything else — an array of numbers, a bare string, a non-JSON body — is wrapped as a single value field, which works but gives later steps nothing to branch on.

The Test panel shows you this

Press Test in the endpoint editor and look at What later steps would read. That is the parsed shape, exactly as a following step will see it. If the field you wanted is not there, fix the function before you wire up the pipeline.

Authentication

Attach a Bearer token credential to the endpoint and Data Conductor sends Authorization: Bearer <token> on every call. Your function checks it:

import os, hmac

def handler(request):
    expected = os.environ["PIPELINE_TOKEN"]
    sent = (request.headers.get("Authorization") or "").removeprefix("Bearer ")
    if not hmac.compare_digest(sent, expected):
        return ({"error": "unauthorized"}, 401)
    ...

Use hmac.compare_digest rather than ==. Generate the token yourself, store it in the function's environment, and paste the same value into the credential — Data Conductor never shows it again.

Work that takes longer than a request

This is the case that needs a pattern, because a step cannot wait. The maximum a single call may take is 120 seconds, and a pipeline step runs once per run — there is no loop that re-calls your function until it says it is finished.

So do not make the caller wait. Return an identifier immediately, and check it in a later step.

Step 1 — start the work, return a handle

def start(request):
    job_id = enqueue_the_real_work()      # returns immediately
    return ({"job_id": job_id, "status": "accepted"}, 200)

Tick Keep the response as pipeline data on this endpoint, so job_id becomes available to the steps after it.

Step 2 — a second endpoint that checks it

Save a second HTTPS endpoint whose URL carries a placeholder:

https://your-function-host/status?job_id={{JOB_ID}}

In the pipeline, declare JOB_ID as an input on that step and bind it to step 1's job_id field. The value is filled in at run time from what step 1 actually returned.

def status(request):
    job_id = request.args["job_id"]
    state = look_up(job_id)               # "running" | "done" | "failed"
    return ({"job_id": job_id, "status": state, "rows": count_for(job_id)}, 200)

Step 3 — act on the answer

Put a condition on the edge out of step 2 and read status. Send done one way and failed another.

There is no waiting, and no retry loop

If step 2 reports running, the run carries on — it does not pause and poll. Design for that:

  • Fast enough to finish? Do the work inside one call and skip all of this. Under 120 seconds, one endpoint is simpler and better.
  • Slow, and the pipeline runs on a schedule? Start it in one run and check it in the next. The identifier is in the previous run's data.
  • Slow, and the answer is needed now? Have your function call back when it finishes, rather than having the pipeline ask.

A tighter loop — the step staying in progress while Data Conductor polls for you — is a feature we have scoped and not built. Ask for it if you need it, and it moves up.

Templates

Each one is the smallest function that satisfies the four rules. All three return a flat object and check a bearer token.

Google Cloud Functions (2nd gen, Python)

import functions_framework, hmac, os

@functions_framework.http
def pipeline_step(request):
    expected = os.environ["PIPELINE_TOKEN"]
    sent = (request.headers.get("Authorization") or "").removeprefix("Bearer ")
    if not hmac.compare_digest(sent, expected):
        return ({"error": "unauthorized"}, 401)

    payload = request.get_json(silent=True) or {}
    result = do_the_work(payload)
    return ({"status": "ok", **result}, 200)

gcloud functions deploy pipeline-step \
  --gen2 --runtime=python312 --region=us-central1 \
  --source=. --entry-point=pipeline_step \
  --trigger-http --no-allow-unauthenticated \
  --set-env-vars PIPELINE_TOKEN=your-generated-token

--no-allow-unauthenticated requires a Google identity as well as your bearer token. Use --allow-unauthenticated if the bearer token is the only check you want — in which case the token is the whole of your security, so generate a long random one.

AWS Lambda (Python, Function URL)

import hmac, json, os

def lambda_handler(event, context):
    expected = os.environ["PIPELINE_TOKEN"]
    sent = (event.get("headers", {}).get("authorization") or "").removeprefix("Bearer ")
    if not hmac.compare_digest(sent, expected):
        return {"statusCode": 401, "body": json.dumps({"error": "unauthorized"})}

    payload = json.loads(event.get("body") or "{}")
    result = do_the_work(payload)
    return {
        "statusCode": 200,
        "headers": {"Content-Type": "application/json"},
        "body": json.dumps({"status": "ok", **result}),
    }

Lambda lower-cases incoming header names on a Function URL, so read authorization, not Authorization. The default Function URL timeout is 30 seconds and the maximum is 15 minutes — but the endpoint's own 120-second cap applies first, so neither helps with slow work. Use the two-step pattern.

Azure Functions (Python v2 model)

import azure.functions as func
import hmac, json, os

app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)

@app.route(route="pipeline-step")
def pipeline_step(req: func.HttpRequest) -> func.HttpResponse:
    expected = os.environ["PIPELINE_TOKEN"]
    sent = (req.headers.get("Authorization") or "").removeprefix("Bearer ")
    if not hmac.compare_digest(sent, expected):
        return func.HttpResponse(json.dumps({"error": "unauthorized"}), status_code=401,
                                 mimetype="application/json")

    payload = req.get_json() if req.get_body() else {}
    result = do_the_work(payload)
    return func.HttpResponse(json.dumps({"status": "ok", **result}),
                             mimetype="application/json")

Azure's own function key arrives as ?code= on the URL. You can use it instead of a bearer token, but put it in the endpoint's URL as a {{FUNCTION_KEY}} variable rather than typing it in — the URL is visible to anyone who can open the endpoint.

Before you wire it into a pipeline

  • Press Test. It issues the request exactly as a run will, with the credential attached, and shows the parsed shape.
  • Check the success statuses. They default to 2xx. If your function answers 202 for accepted work, that is already covered; if it answers 404 to mean "gone, as expected", add 404 so the step does not fail.
  • Give it a name a stranger can read. It appears in the step picker, and the person adding it to a pipeline in six months may not be you.