AI Inventory Demand Forecasting with Kestra

id: ai-inventory-forecasting
namespace: company.team

inputs:
  - id: inventory_api_url
    type: STRING
    displayName: Inventory API URL
    description: Endpoint returning current warehouse inventory levels as JSON
    defaults: "https://api.yourwarehouse.com/v1/inventory"
  - id: sales_api_url
    type: STRING
    displayName: Sales velocity API URL
    description: Endpoint returning recent sales data as JSON
    defaults: "https://api.yourwarehouse.com/v1/sales/velocity"
  - id: supplier_api_url
    type: STRING
    displayName: Supplier PO API URL
    description: Endpoint to POST new purchase orders to the supplier
    defaults: "https://api.yoursupplier.com/v1/orders"
  - id: erp_api_url
    type: STRING
    displayName: ERP API URL
    description: ERP system endpoint for logging confirmed purchase orders
    defaults: "https://erp.yourcompany.com/api/purchase-orders"

tasks:
  - id: fetch_data
    type: io.kestra.plugin.core.flow.Parallel
    tasks:
      - id: fetch_inventory
        type: io.kestra.plugin.core.http.Request
        uri: "{{ inputs.inventory_api_url }}"
        method: GET
        headers:
          Authorization: "Bearer {{ secret('WAREHOUSE_API_TOKEN') }}"

      - id: fetch_sales_velocity
        type: io.kestra.plugin.core.http.Request
        uri: "{{ inputs.sales_api_url }}"
        method: GET
        headers:
          Authorization: "Bearer {{ secret('WAREHOUSE_API_TOKEN') }}"

  - id: ai_demand_forecast
    type: io.kestra.plugin.openai.ChatCompletion
    apiKey: "{{ secret('OPENAI_API_KEY') }}"
    model: gpt-4o
    maxTokens: 1500
    jsonResponseSchema: |
      {
        "type": "object",
        "properties": {
          "items": {
            "type": "array",
            "items": {
              "type": "object",
              "properties": {
                "sku": {"type": "string"},
                "current_stock": {"type": "number"},
                "predicted_demand_7d": {"type": "number"},
                "reorder_needed": {"type": "boolean"},
                "recommended_quantity": {"type": "number"},
                "confidence": {"type": "number", "minimum": 0, "maximum": 1},
                "reasoning": {"type": "string"}
              },
              "required": ["sku", "reorder_needed", "recommended_quantity", "confidence"]
            }
          }
        },
        "required": ["items"]
      }
    messages:
      - role: system
        content: |
          You are a supply chain AI analyst. Given warehouse inventory levels and sales velocity data,
          predict demand for the next 7 days and identify which items need reordering.

          For each item, compute:
          - Predicted demand over the next 7 days based on sales velocity trends
          - Whether reordering is needed (stock will run out within 7 days)
          - Recommended reorder quantity (2-week safety stock)
          - Confidence score (0-1) in the forecast
          - Brief reasoning

          Return structured JSON only.
      - role: user
        content: |
          Inventory data: {{ outputs.fetch_data.fetch_inventory.body }}
          Sales velocity: {{ outputs.fetch_data.fetch_sales_velocity.body }}

  - id: filter_and_create_orders
    type: io.kestra.plugin.scripts.python.Script
    inputFiles:
      forecast.json: "{{ outputs.ai_demand_forecast.choices[0].message.content }}"
    script: |
      import json
      from kestra import Kestra

      with open("forecast.json") as f:
          forecast = json.load(f)

      items_needing_reorder = [
          item for item in forecast.get("items", [])
          if item.get("reorder_needed") and item.get("recommended_quantity", 0) > 0
      ]

      purchase_orders = []
      for item in items_needing_reorder:
          po = {
              "sku": item["sku"],
              "quantity": item["recommended_quantity"],
              "confidence": item["confidence"],
              "reasoning": item.get("reasoning", ""),
              "po_number": f"PO-{item['sku']}-{{ now() | date('yyyyMMdd-HHmm') }}"
          }
          purchase_orders.append(po)

      Kestra.outputs({
          "purchase_orders": purchase_orders,
          "reorder_count": len(purchase_orders),
          "has_orders": len(purchase_orders) > 0
      })

  - id: process_orders
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.filter_and_create_orders.vars.has_orders }}"
    then:
      - id: submit_orders
        type: io.kestra.plugin.core.flow.ForEach
        values: "{{ outputs.filter_and_create_orders.vars.purchase_orders }}"
        concurrencyLimit: 3
        tasks:
          - id: send_to_supplier
            type: io.kestra.plugin.core.http.Request
            uri: "{{ inputs.supplier_api_url }}"
            method: POST
            contentType: application/json
            headers:
              Authorization: "Bearer {{ secret('SUPPLIER_API_TOKEN') }}"
            body: |
              {
                "po_number": "{{ taskrun.value | jq('.po_number') | first }}",
                "sku": "{{ taskrun.value | jq('.sku') | first }}",
                "quantity": {{ taskrun.value | jq('.quantity') | first }},
                "notes": "AI forecast (confidence: {{ taskrun.value | jq('.confidence') | first }}): {{ taskrun.value | jq('.reasoning') | first }}"
              }

      - id: log_to_postgres
        type: io.kestra.plugin.jdbc.postgresql.Queries
        fetchType: NONE
        sql: |
          INSERT INTO purchase_orders (po_number, sku, quantity, forecast_confidence, reasoning, created_at)
          SELECT
            po->>'po_number',
            po->>'sku',
            (po->>'quantity')::numeric,
            (po->>'confidence')::numeric,
            po->>'reasoning',
            NOW()
          FROM json_array_elements('{{ outputs.filter_and_create_orders.vars.purchase_orders | json }}'::json) AS po
          ON CONFLICT (po_number) DO NOTHING;

      - id: notify_team
        type: io.kestra.plugin.email.MailSend
        from: "{{ secret('ALERT_EMAIL_FROM') }}"
        to: "{{ secret('ALERT_EMAIL_TO') }}"
        subject: "[Supply Chain] {{ outputs.filter_and_create_orders.vars.reorder_count
          }} Purchase Orders Generated"
        htmlTextContent: |
          <h2>Automated Purchase Orders, AI Inventory Forecast</h2>
          <p><strong>{{ outputs.filter_and_create_orders.vars.reorder_count }}</strong> items flagged for reorder.</p>
          <pre>{{ outputs.filter_and_create_orders.vars.purchase_orders | json }}</pre>
          <p><em>Generated automatically by Kestra AI Inventory Forecasting workflow.</em></p>
        host: "{{ secret('SMTP_HOST') }}"
        username: "{{ secret('SMTP_USERNAME') }}"
        password: "{{ secret('SMTP_PASSWORD') }}"
        port: 465

pluginDefaults:
  - type: io.kestra.plugin.jdbc.postgresql.Queries
    values:
      url: "{{ secret('POSTGRES_URL') }}"
      username: "{{ secret('POSTGRES_USER') }}"
      password: "{{ secret('POSTGRES_PASSWORD') }}"

triggers:
  - id: every_6_hours
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 */6 * * *"
    description: Run every 6 hours to monitor inventory and sales trends

Automate supply chain replenishment end to end. This blueprint pulls live warehouse inventory and sales velocity every 6 hours, sends both to OpenAI GPT-4o for a structured 7-day demand forecast, then automatically generates purchase orders for SKUs at risk of stockout, submits them to your supplier, logs them to PostgreSQL, and emails your team a digest. It replaces manual stockout monitoring and static reorder points with AI-driven demand prediction, helping retail, e-commerce, and manufacturing teams cut both stockouts and excess inventory.

How it works

A io.kestra.plugin.core.trigger.Schedule trigger fires on cron 0 */6 * * *, running the flow every 6 hours.
fetch_data, a io.kestra.plugin.core.flow.Parallel task, runs two io.kestra.plugin.core.http.Request calls at once: fetch_inventory and fetch_sales_velocity, both authenticated with a bearer token.
ai_demand_forecast (io.kestra.plugin.openai.ChatCompletion, model gpt-4o) receives the combined data and returns a JSON-Schema-constrained forecast with predicted_demand_7d, reorder_needed, recommended_quantity, confidence, and reasoning per SKU.
filter_and_create_orders, a io.kestra.plugin.scripts.python.Script, keeps only items where reorder_needed is true and builds purchase orders with generated PO numbers, emitting them via Kestra.outputs.
process_orders, a io.kestra.plugin.core.flow.If, runs only when has_orders is true. Inside, a io.kestra.plugin.core.flow.ForEach (concurrencyLimit: 3) POSTs each order to the supplier API, log_to_postgres upserts them with io.kestra.plugin.jdbc.postgresql.Queries, and notify_team sends a summary via io.kestra.plugin.email.MailSend.

What you get

A hands-off reorder loop running every 6 hours
GPT-4o forecasts returned as validated JSON via jsonResponseSchema
Parallel supplier submission with bounded concurrency
An idempotent PostgreSQL audit log (ON CONFLICT DO NOTHING)
An HTML email digest with quantities and AI confidence scores

Who it's for

Retail and e-commerce teams automating replenishment
Manufacturing operations balancing raw material stock against production
Supply chain and procurement teams replacing rule-based reorder points with AI

Why orchestrate this with Kestra

OpenAI has no scheduler, no retry logic, and no way to gate downstream supplier writes on a forecast result. Kestra supplies the event trigger, parallel and conditional control flow, per-task retries, and full execution lineage across HTTP, AI, Python, JDBC, and email steps. The entire pipeline is declarative YAML you can version, review, and replay, with every input, output, and forecast captured for audit.

Prerequisites

A Kestra instance (0.18.0+ recommended)
Reachable inventory, sales, supplier, and ERP API endpoints
A PostgreSQL database with a purchase_orders table
An OpenAI API key with gpt-4o access

Secrets

WAREHOUSE_API_TOKEN: bearer token for the inventory and sales APIs
OPENAI_API_KEY: OpenAI API key for GPT-4o
SUPPLIER_API_TOKEN: bearer token for the supplier purchase order API
POSTGRES_URL, POSTGRES_USER, POSTGRES_PASSWORD: PostgreSQL connection
ALERT_EMAIL_FROM, ALERT_EMAIL_TO: email sender and recipient
SMTP_HOST, SMTP_USERNAME, SMTP_PASSWORD: SMTP server credentials

Quick start

Add the blueprint to your namespace and define the secrets above.
Set the four input URLs (inventory, sales, supplier, ERP) to your endpoints.
Create the purchase_orders table with a unique constraint on po_number.
Run once manually to confirm the forecast and order logic, then let the schedule take over.

How to extend

Add a human-in-the-loop Pause task to approve orders before submission.
Push confirmed orders into your ERP (SAP, Oracle, NetSuite) via its REST API.
Add a Slack alert for high-confidence recommendations.
Swap or tune the model, or branch on confidence to auto-approve only above a threshold.

AI-Powered Inventory Demand Forecasting, Auto-Generate Purchase Orders with OpenAI

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

AI-Powered Inventory Demand Forecasting, Auto-Generate Purchase Orders with OpenAI

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Quick start

How to extend

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More AI Blueprints

Trigger the Right Automation with an AI Agent Based on User Intent

Convert Natural Language Prompts to Structured Todoist Tasks with Deepseek AI

AI Document Data Extraction: Parse PDFs and Web Pages into Structured JSON with OpenAI

Generate a book chapter summary using Perplexity AI and email it.

Draft GDPR Breach Notifications for Supervisory Authorities and Data Subjects

Decide When to Take Action Using AI, Check Real-Time Weather and Trigger Automations

More Data Blueprints

Generate and Execute SQL Queries from Natural Language Using OpenAI GPT-4o

Build an ETL Pipeline with Airbyte Cloud and dbt Core for SaaS Analytics

Sync Postgres records to Algolia index

Build an ETL Pipeline with Airbyte Cloud and dbt Cloud for SaaS Analytics

AI Document Data Extraction: Parse PDFs and Web Pages into Structured JSON with OpenAI

Trigger an Airflow DAG

New to Kestra?