https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> How can I successfully pass {{flow_ru...
# ask-marvin
b
@Marvin How can I successfully pass {{flow_run}} json to a custom webhook via an automation?
m
✅ thought for 69.8 seconds
Based on the research, I can provide you with a detailed explanation of how to pass flow run JSON to a custom webhook via an automation. Here's how to do it: 1. First, create a CustomWebhookNotificationBlock in your Prefect workspace:
Copy code
python
from prefect.blocks.notifications import CustomWebhookNotificationBlock

webhook_block = CustomWebhookNotificationBlock(
    name="my-custom-webhook",
    url="YOUR_WEBHOOK_URL",
    method="POST",  # Default is POST, but you can change if needed
    headers={"Content-Type": "application/json"}
)
webhook_block.save()
2. In the Prefect UI, create a new automation with: - A trigger condition (e.g., "When a flow run enters a Failed state") - An action using the "Call Webhook" action type 3. When configuring the webhook in the automation, you can use the following template variables in your JSON payload:
Copy code
json
{
    "flow_run_data": {{flow_run}},
    "flow_name": "{{flow_run.flow_name}}",
    "flow_run_id": "{{flow_run.id}}",
    "flow_run_status": "{{flow_run.state.type}}",
    "flow_run_message": "{{flow_run.state.message}}"
}
Key points to remember: - The 
{{flow_run}}
template variable contains the entire flow run object - You can access specific fields using dot notation: 
{{flow_run.field_name}}
- The webhook will automatically parse these template variables before sending - Make sure your receiving endpoint can handle the JSON payload structure you define Common fields available in the 
flow_run
object include: - `id`: The unique identifier of the flow run - `flow_name`: Name of the flow - `state.type`: Current state of the flow run (e.g., "COMPLETED", "FAILED") - `state.message`: Any message associated with the state - `start_time`: When the flow run started - `end_time`: When the flow run ended For example, if you want to send a notification to Slack with flow run details, your webhook payload might look like:
Copy code
json
{
    "text": "Flow Run Update",
    "blocks": [
        {
            "type": "section",
            "text": {
                "type": "mrkdwn",
                "text": "Flow *{{flow_run.flow_name}}* (ID: {{flow_run.id}}) is now in state *{{flow_run.state.type}}*\nMessage: {{flow_run.state.message}}"
            }
        }
    ]
}
Remember that the receiving webhook endpoint needs to be able to handle the JSON structure you send. Make sure to test the webhook with a sample payload before setting up the automation. Links for further reading: - Prefect Automations Documentation - Prefect Webhooks Guide Would you like me to provide more specific examples for your use case or clarify any part of the setup process?
b
Copy code
@Marvin     "reason": "The notification block was invalid: KeyError(\"json_data/Placeholder(full_match='{{flow_run}}', name='flow_run', type=<PlaceholderType.STANDARD: 'standard'>)\")",
@Marvin
Copy code
"reason": "The notification block was invalid: KeyError(\"json_data/Placeholder(full_match='{{flow_run}}', name='flow_run', type=<PlaceholderType.STANDARD: 'standard'>)\")",
m
✅ thought for 56.4 seconds
Based on the error you're seeing and my research, I can now provide the correct way to handle this. The issue is with how the 
{{flow_run}}
variable is being passed in the JSON data. Here's how to fix it: Instead of passing the flow run directly as a raw template variable, you need to access specific attributes of the flow run object. Here are two approaches: 1. Using specific fields (Recommended):
Copy code
json
{
    "flow_run_id": "{{flow_run.id}}",
    "flow_name": "{{flow_run.flow_name}}",
    "state": "{{flow_run.state.type}}",
    "message": "{{flow_run.state.message}}",
    "start_time": "{{flow_run.start_time}}",
    "end_time": "{{flow_run.end_time}}"
}
2. Using the built-in 
to_json
filter:
Copy code
json
{
    "flow_run_data": "{{flow_run | to_json}}"
}
The error you're seeing occurs because the webhook system can't directly serialize the 
{{flow_run}}
placeholder into valid JSON. By either: - Breaking it down into specific fields you want to access, or - Using the 
to_json
filter You should be able to resolve the error. When setting up your CustomWebhookNotificationBlock, make sure to structure it like this:
Copy code
python
from prefect.blocks.notifications import CustomWebhookNotificationBlock

webhook_block = CustomWebhookNotificationBlock(
    name="my-custom-webhook",
    url="YOUR_WEBHOOK_URL",
    method="POST",
    headers={"Content-Type": "application/json"},
    json_data={
        "flow_run_id": "{{flow_run.id}}",
        "flow_name": "{{flow_run.flow_name}}",
        "state": "{{flow_run.state.type}}",
        "message": "{{flow_run.state.message}}"
    }
)
webhook_block.save()
Would you like to share what specific data you're trying to capture from the flow run? That way, I can provide a more tailored example for your use case.
b
@Marvin in the automation, when i do {{flow_run|to_json}} for the body i get "body": "{\"flow_run\":\"Failed to render template due to the following error: TemplateAssertionError(\\\"No filter named 'to_json'.\\\")\\nTemplate source:\\n{{flow_run|to_json}}\"}",
m
✅ thought for 32.6 seconds
Based on this research, let me provide the correct approach. Instead of using the 
to_json
filter, you'll need to access the flow run properties directly. Here's how you can structure the webhook body in your automation:
Copy code
json
{
    "id": "{{flow_run.id}}",
    "name": "{{flow_run.name}}",
    "flow_name": "{{flow_run.flow_name}}",
    "state_type": "{{flow_run.state.type}}",
    "state_message": "{{flow_run.state.message}}",
    "start_time": "{{flow_run.start_time}}",
    "end_time": "{{flow_run.end_time}}",
    "parameters": "{{flow_run.parameters}}",
    "tags": "{{flow_run.tags}}",
    "deployment_id": "{{flow_run.deployment_id}}"
}
If you need to include specific nested data or have particular fields you want to capture, please let me know and I can help you structure the JSON appropriately. Also, if you're trying to debug what data is available, you could try setting up a simple webhook (like webhook.site) and first send a simpler payload to see what data you have access to:
Copy code
json
{
    "debug_info": "Flow run {{flow_run.id}} completed with state {{flow_run.state.type}}"
}
Would you like to share what specific flow run data you're trying to capture? That way, I can help you structure the JSON payload more precisely for your needs.
Open in Slack
PreviousNext
Powered by