Doug Applegate
12/10/2025, 6:44 PMMarvin
12/10/2025, 6:44 PMMarvin
12/10/2025, 6:47 PMimport hmac, hashlib, os, requests
from fastapi import FastAPI, Request, HTTPException
app = FastAPI()
# Set these in your environment
PREFECT_WEBHOOK_URL = os.environ["PREFECT_WEBHOOK_URL"] # Your Prefect Cloud webhook URL
PREFECT_API_KEY = os.environ["PREFECT_API_KEY"] # Service account API key
WEBHOOK_SECRET = os.environ["WEBHOOK_SECRET"] # Shared secret for HMAC
def verify_signature(signature_header: str, body: bytes) -> bool:
# Expecting: "sha256=abcdef1234..."
try:
algo, provided = signature_header.split("=", 1)
except ValueError:
return False
if algo.lower() != "sha256":
return False
mac = hmac.new(WEBHOOK_SECRET.encode(), msg=body, digestmod=hashlib.sha256)
expected = mac.hexdigest()
return hmac.compare_digest(provided, expected)
@app.post("/inbound")
async def inbound(request: Request):
body = await request.body()
sig = request.headers.get("X-Hub-Signature-256") or request.headers.get("X-Signature")
if not sig or not verify_signature(sig, body):
raise HTTPException(status_code=401, detail="Invalid signature")
# Forward to Prefect webhook with service account auth
headers = {
"Authorization": f"Bearer {PREFECT_API_KEY}",
"Content-Type": request.headers.get("Content-Type", "application/json"),
}
resp = <http://requests.post|requests.post>(PREFECT_WEBHOOK_URL, data=body, headers=headers, timeout=10)
if not resp.ok:
raise HTTPException(status_code=502, detail=f"Prefect webhook error: {resp.status_code} {resp.text}")
return {"status": "ok"}
Notes
- Prefect Cloud inbound: no built-in X-Signature/HMAC validation; use service account auth if possible.
- Prefect outbound (if you meant validating webhooks that Prefect sends to you): Prefect can send custom headers, but it does not compute HMAC signatures for you. You can include static headers, but there’s no built-in signing. Your receiver should rely on a shared secret header or implement its own validation.
Docs:
- Webhook concepts: Prefect webhooks
- Create/manage a webhook: How to create a webhook
- API reference: Prefect Cloud API docs
Does this match your use case? Are you trying to validate requests into Prefect (inbound) or validate requests Prefect sends to your system (outbound)?Doug Applegate
12/10/2025, 7:07 PMMarvin
12/10/2025, 7:08 PMMarvin
12/10/2025, 7:11 PMDoug Applegate
12/10/2025, 7:13 PMMarvin
12/10/2025, 7:14 PMMarvin
12/10/2025, 7:17 PMheaders, body, and method.
- The Authorization header is stripped for security; other headers (e.g., X-Signature, User-Agent) are available.
- Whatever you place into the webhook event’s payload (or resource) becomes available to your Automation action templates (e.g., RunDeployment parameters).
Example: capture headers in the webhook template
{
"event": "external.webhook.received",
"resource": {
"prefect.resource.id": "webhook.{{ headers['X-Source'] | default('unknown') }}"
},
"payload": {
"signature": "{{ headers['X-Signature'] | default('') }}",
"user_agent": "{{ headers['User-Agent'] | default('') }}",
// If you want many headers, you can include them as a JSON object:
"all_headers": {{ headers | tojson }}
}
}
Then pass them to your deployment in an Automation (RunDeployment action)
- In the UI, set deployment parameters using Jinja templates like:
- {{ event.payload.signature }}
- {{ event.payload.user_agent }}
- For specific items from the bundled headers: {{ event.payload.all_headers['X-Source'] }}
Example via Python SDK
from datetime import timedelta
from prefect.automations import Automation
from prefect.events.schemas.automations import EventTrigger, Posture
from prefect.events.actions import RunDeployment
automation = Automation(
name="Webhook → Deployment",
trigger=EventTrigger(
expect={"external.webhook.received"},
posture=Posture.Reactive,
threshold=1,
within=timedelta(0),
),
actions=[
RunDeployment(
source="selected",
deployment_id="<your-deployment-id>",
parameters={
"signature": {"__prefect_kind": "jinja", "template": "{{ event.payload.signature }}"},
"user_agent": {"__prefect_kind": "jinja", "template": "{{ event.payload.user_agent }}"},
// If you captured many headers:
// "headers_json": {"__prefect_kind": "jinja", "template": "{{ event.payload.all_headers | tojson }}"},
},
)
],
).create()
Docs
- Webhook concepts and templating: Webhooks
- Create/manage a webhook: Create a webhook
- Automations: Automations
If you can share which headers you need (e.g., X-Signature, X-Request-ID), I can suggest the exact template and parameter mapping.