JSON API Response Validator

id: json-api-response-validator
namespace: company.team

description: |
  Call a configurable HTTP API, validate its JSON response against a small
  contract, and route valid and invalid responses independently.

triggers:
  - id: scheduled_validation
    type: io.kestra.plugin.core.trigger.Schedule
    description: Run the API validation on a schedule. Disabled until configured.
    cron: "0 * * * *"
    disabled: true

inputs:
  - id: api_url
    type: STRING
    displayName: API URL
    description: Fully-qualified HTTP or HTTPS URL that returns the JSON document to
      validate.
    defaults: https://dummyjson.com/products/1
    required: true

  - id: expected_status_code
    type: INT
    displayName: Expected HTTP status
    description: HTTP status code required for the response to be considered valid.
    defaults: 200
    required: true

  - id: required_fields
    type: JSON
    displayName: Required JSON fields
    description: JSON array of top-level field names that must be present in the
      response object.
    defaults: |
      ["id", "title", "price"]
    required: true

  - id: notify_slack
    type: BOOL
    displayName: Send invalid-response Slack notification
    description: When true, send invalid results to Slack using the
      SLACK_WEBHOOK_URL secret.
    defaults: false
    required: true

tasks:
  - id: call_api
    type: io.kestra.plugin.core.http.Request
    description: Call the configured API and retain unsuccessful HTTP responses for
      validation.
    uri: "{{ inputs.api_url }}"
    method: GET
    headers:
      Accept: application/json
    options:
      allowFailed: true

  - id: validate_response
    type: io.kestra.plugin.scripts.python.Script
    description: Validate the status code, JSON object shape, and required top-level fields.
    taskRunner:
      type: io.kestra.plugin.core.runner.Process
    env:
      RESPONSE_BODY: "{{ outputs.call_api.body }}"
      RESPONSE_STATUS: "{{ outputs.call_api.statusCode }}"
      EXPECTED_STATUS: "{{ inputs.expected_status_code }}"
      REQUIRED_FIELDS: "{{ inputs.required_fields | toJson }}"
    script: |
      import json
      import os

      expected_status = int(os.environ["EXPECTED_STATUS"])
      actual_status = int(os.environ["RESPONSE_STATUS"])
      required_fields = json.loads(os.environ["REQUIRED_FIELDS"])
      response_body = os.environ.get("RESPONSE_BODY", "")

      errors = []
      parsed_response = None
      if actual_status != expected_status:
          errors.append(
              f"expected HTTP {expected_status}, received HTTP {actual_status}"
          )

      try:
          parsed_response = json.loads(response_body)
      except json.JSONDecodeError:
          errors.append("response body is not valid JSON")

      if parsed_response is not None and not isinstance(parsed_response, dict):
          errors.append("response JSON must be an object")

      missing_fields = []
      if isinstance(parsed_response, dict):
          missing_fields = [
              field for field in required_fields if field not in parsed_response
          ]
          if missing_fields:
              errors.append(
                  "missing required fields: " + ", ".join(missing_fields)
              )

      result = {
          "valid": not errors,
          "http_status": actual_status,
          "expected_status": expected_status,
          "missing_fields": missing_fields,
          "errors": errors,
      }
      print("::" + json.dumps({"outputs": {"result": result}}) + "::")

  - id: route_validation_result
    type: io.kestra.plugin.core.flow.If
    description: Route valid and invalid API responses to separate handling paths.
    condition: "{{ outputs.validate_response.vars.result.valid }}"
    then:
      - id: log_valid_response
        type: io.kestra.plugin.core.log.Log
        message: |
          API response is valid.
          URL: {{ inputs.api_url }}
          HTTP status: {{ outputs.validate_response.vars.result.http_status }}
    else:
      - id: log_invalid_response
        type: io.kestra.plugin.core.log.Log
        message: |
          API response is invalid.
          URL: {{ inputs.api_url }}
          Validation errors: {{ outputs.validate_response.vars.result.errors }}

      - id: notify_invalid_response
        type: io.kestra.plugin.core.flow.If
        description: Send an optional Slack notification without exposing API credentials.
        condition: "{{ inputs.notify_slack }}"
        then:
          - id: send_slack_notification
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Notify the team when the API response does not satisfy the
              contract.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            messageText: |
              JSON API validation failed for {{ inputs.api_url }}.
              HTTP status: {{ outputs.validate_response.vars.result.http_status }}
              Errors: {{ outputs.validate_response.vars.result.errors }}

outputs:
  - id: validation_result
    type: JSON
    description: Validation result containing validity, status code, missing fields,
      and errors.
    value: "{{ outputs.validate_response.vars.result | toJson }}"

Validate an API response as part of a repeatable Kestra workflow. The blueprint calls a configurable URL, accepts unsuccessful HTTP responses as data for inspection, parses the response as JSON, and checks a small top-level contract. It then uses an If flow to handle valid and invalid responses and returns the validation result as a structured flow output.

How it works

The scheduled_validation trigger is provided as a disabled hourly schedule. Enable it after configuring the target API.
call_api uses io.kestra.plugin.core.http.Request to send a GET request to the api_url input. options.allowFailed: true keeps 4xx and 5xx responses available for validation rather than failing before the contract check.
validate_response parses the response body and checks that the HTTP status equals expected_status_code, the body is a JSON object, and every field in required_fields exists.
route_validation_result uses an If flow. Valid responses are logged as successful; invalid responses are logged with their validation errors and can optionally be sent to Slack.
The validation_result output returns the status, validity flag, missing fields, and validation errors.

What you get

A configurable API URL and expected HTTP status.
Explicit handling of unsuccessful HTTP responses.
JSON parsing and required-field validation.
Separate valid and invalid response paths.
A structured validation_result output for downstream tasks.
Optional Slack notifications using a Kestra secret rather than a hardcoded credential.

Who it's for

QA engineers testing API contracts.
Platform and SRE teams building synthetic API checks.
Developers validating integrations before downstream processing.

Why orchestrate this with Kestra

A standalone script can make one request, but Kestra adds triggers, execution history, observable task boundaries, branching, secret management, and reusable outputs. The same validation can run manually, on a schedule, or as part of a larger data and integration workflow.

Prerequisites

A Kestra instance with the Core HTTP and Python Script plugins available.
Network access from the Kestra worker to the configured API URL.
A JSON API response matching the expected contract.

Secrets

SLACK_WEBHOOK_URL: Slack Incoming Webhook URL, required only when notify_slack is true. Do not put the webhook URL in this flow.

Expected response structure

With the default inputs, a valid response looks like:

{
  "id": 1,
  "title": "Example product",
  "price": 29.99
}

The response must be a JSON object with id, title, and price, and it must use HTTP status 200. Change required_fields and expected_status_code for another API contract.

Quick start

Import this flow and execute it manually with the default inputs.
Set api_url to the API endpoint you want to validate.
Set required_fields to a JSON array of required top-level fields.
Set notify_slack to true only after creating the SLACK_WEBHOOK_URL secret in Kestra.
Review {{ outputs.validation_result }} and enable the schedule when the manual execution succeeds.

Validation conditions

A response is valid only when its HTTP status equals expected_status_code, its body parses as JSON, the parsed value is an object, and every name in required_fields is present. A 4xx/5xx status, malformed JSON, an array/scalar body, or a missing field produces an invalid result with a reason in errors.

Outputs

{{ outputs.validation_result }}: JSON object containing valid, http_status, expected_status, missing_fields, and errors.
{{ outputs.validate_response.vars.result }}: The same structured result available to downstream tasks in this flow.

How to extend

Add nested JSON-path checks for values such as status or data.items.
Add POST request inputs and a JSON request body for contract tests.
Add email, PagerDuty, or another notification plugin to the invalid branch.
Replace the hourly schedule with a webhook or upstream flow trigger.

JSON API Response Validator

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Expected response structure

Quick start

Validation conditions

Outputs

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

JSON API Response Validator

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Expected response structure

Quick start

Validation conditions

Outputs

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Load time series data into InfluxDB

Idempotent Incremental CSV Loader from S3 to PostgreSQL

Huawei GeminiDB TTL Hygiene Sentinel

Huawei DLI Data Quality Assertion Gate

InfluxDB Bucket Retention Sentinel

Idempotent API Poller with Dynamic KV Watermark

More Core Blueprints

Govern cloud infrastructure with Kestra Assets: automated S3 bucket provisioning by team

Block a release when k6 shows it is slower than the last good one

HTTP Redirect Chain Audit

Idempotent API Poller with Dynamic KV Watermark

Snapshot Pages to the Wayback Machine After Deploy

Automated Incident Response Threat Intelligence Alerting

New to Kestra?