Automating Product Notifications with PostgreSQL Triggers and Langflow Agents

Situation

I wanted to automate product availability notifications to customers based on their expressed interest. Manually notifying users every time a product became available can be inefficient, especially when dealing with large customer databases and personalized messages.

Get Dreary’s stories in your inbox

Join Medium for free to get updates from this writer.

Remember me for faster sign in

The broader context was exploring Langflow as a no-code/low-code orchestration layer to handle intelligent agent collaboration in real-world automation scenarios, integrating with existing tools like PostgreSQL, Gmail, and Google Sheets.

Task

My goal was to:

• Detect when a product becomes available.
• Notify all interested customers automatically via email.
• Personalize messages based on customer type (e.g., VIP vs standard).
• Log data into Google Sheets for visualization in Grafana.

Action

I used a PostgreSQL trigger to listen for changes in product availability. When a product’s available field changed to true, a function collected all interested customer emails and their status (VIP or standard) and emitted a JSON payload via pg_notify.

A Python script (webhook listener) received the notification, parsed the payload, and forwarded it to a Langflow webhook endpoint.

import asyncio
import psycopg2
import psycopg2.extensions
import aiohttp
import json

# DB configuration (should use env variables...)
DB_CONFIG = {
    "database": "<db_name>",
    "user": "<db_user>",
    "password": "<db_password>",
    "host": "<host_ip>",
    "port": "<port>"
}
                                                    # flow id here 
WEBHOOK_URL = "http://127.0.0.1:7860/api/v1/webhook/<langflow_flow_id>" 
CHANNEL = "webhook_channel"

# DB connection
def init_db_connection():
    conn = psycopg2.connect(**DB_CONFIG)
    conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
    cursor = conn.cursor()
    cursor.execute(f"LISTEN {CHANNEL};")
    return conn


# Send async webhook: this is responsible for sending the payload received from postgreSQL 
# to the langflow flow specified above (WEBHOOK_URL)
async def send_webhook(payload_data):
    async with aiohttp.ClientSession() as session:
        try:
            async with session.post(WEBHOOK_URL, json=payload_data) as resp:
                print(f"Webhook sent (status {resp.status}) with data:\n{json.dumps(payload_data, indent=2)}")
        except aiohttp.ClientError as e:
            print(f"Failed to send webhook: {e}")

# Listener 
async def listen_for_notifications():
    conn = init_db_connection()
    print(f"Listening on PostgreSQL channel: {CHANNEL}")

    while True:
        conn.poll()
        while conn.notifies:
            notification = conn.notifies.pop(0)
            print(f"[RECEIVED from DB] Raw Payload:\n{notification.payload}\n")

            try:
                payload_data = json.loads(notification.payload)
                await send_webhook(payload_data)
            except json.JSONDecodeError as e:
                print(f"Invalid JSON from trigger: {e}")

        await asyncio.sleep(1)  # evita busy-waiting

# Launch
if __name__ == "__main__":
    asyncio.run(listen_for_notifications())

Within Langflow:

• I built a multi-agent architecture with an orchestrator agent.
• The orchestrator parsed the payload and dispatched the task to two specialized agents:
  - vip_email_agent
  - standard_email_agent

Both used Composio’s Gmail tool to generate customized email drafts.

• A third agent, make_google_sheet_agent, used a custom Langflow block to perform an HTTP POST request to a Make.com webhook, inserting the data into Google Sheets.

Finally, I connected Grafana to the Google Sheet to visualize product interest activity.

Langflow flow

Result

• The system can now automatically detect product availability and prepare targeted email drafts without manual steps;

Emails generated by the agents

• All data is logged to a centralized sheet

and visualized live in a Grafana dashboard.

• The workflow is modular and extendable, making it easy to plug in other agents or outputs.