Implementation Guide: Monitor client bank feeds and surface cash flow alerts proactively

Hardware Procurement

Software Procurement

Prerequisites

• Client accounting firm must have active QuickBooks Online Plus/Advanced or Xero Growing/Established subscriptions for all monitored client entities, with bank feeds already connected and reconciling
• Client must provide a designated 'Qualified Individual' per FTC Safeguards Rule who will be the compliance point of contact and approve AI system access to financial data
• Client must have a Written Information Security Plan (WISP) per IRS Publication 4557 — if not, MSP should scope a WISP creation engagement as a prerequisite project
• All accounting staff who will receive and act on alerts must have email addresses and access to the chosen notification platform (Slack, Microsoft Teams, or email)
• Client must provide OAuth 2.0 authorization for each QBO or Xero organization to be monitored — this requires a user with Admin or Accountant role in each entity
• MSP must have an Intuit Developer account (free registration at developer.intuit.com) and/or Xero Developer account (free registration at developer.xero.com) with production app credentials
• MSP must have an OpenAI Platform account with billing enabled and API keys generated (platform.openai.com)
• Azure subscription (or equivalent cloud provider account) with permissions to create VMs, networking resources, and managed databases
• Domain name or subdomain available for the n8n instance (e.g., n8n.mspdomain.com) with ability to create DNS A records and obtain SSL certificates
• Client must define initial cash flow alert thresholds for each monitored entity: minimum cash balance, large transaction threshold (e.g., >$5,000), and expected revenue ranges
• Network firewall must allow outbound HTTPS (port 443) to: api.intuit.com, api.xero.com, api.openai.com, floatapp.com, hooks.slack.com, and the n8n webhook domain
• MFA must be enabled on all accounts that access financial data: cloud hosting, QBO/Xero admin, OpenAI, n8n admin, and database connections
• Client must provide a list of all entities to be monitored with: entity name, accounting platform (QBO or Xero), primary contact, and industry vertical for alert tuning
• Backup infrastructure must be in place: either the Synology NAS (on-prem) or Azure Blob Storage for nightly database and configuration backups

Installation Steps

Step 1: Provision Azure Infrastructure

Create the production and staging virtual machines, networking resources, and managed PostgreSQL database in Azure. Use the East US 2 or closest region to the client. Configure a Virtual Network with a Network Security Group allowing only inbound SSH (port 22 from MSP IP range) and HTTPS (port 443 from anywhere). Attach a static public IP to the production VM.

az group create --name rg-cashflow-agent --location eastus2
az network vnet create --resource-group rg-cashflow-agent --name vnet-cashflow --address-prefix 10.0.0.0/16 --subnet-name snet-agent --subnet-prefix 10.0.1.0/24
az network nsg create --resource-group rg-cashflow-agent --name nsg-agent
az network nsg rule create --resource-group rg-cashflow-agent --nsg-name nsg-agent --name AllowSSH --priority 100 --direction Inbound --access Allow --protocol Tcp --source-address-prefixes <MSP_IP_RANGE> --destination-port-ranges 22
az network nsg rule create --resource-group rg-cashflow-agent --nsg-name nsg-agent --name AllowHTTPS --priority 110 --direction Inbound --access Allow --protocol Tcp --source-address-prefixes '*' --destination-port-ranges 443
az network public-ip create --resource-group rg-cashflow-agent --name pip-agent-prod --sku Standard --allocation-method Static
az vm create --resource-group rg-cashflow-agent --name vm-agent-prod --image Canonical:ubuntu-24_04-lts:server:latest --size Standard_B2ms --admin-username azureadmin --ssh-key-value ~/.ssh/id_rsa.pub --vnet-name vnet-cashflow --subnet snet-agent --nsg nsg-agent --public-ip-address pip-agent-prod --os-disk-size-gb 64
az vm create --resource-group rg-cashflow-agent --name vm-agent-staging --image Canonical:ubuntu-24_04-lts:server:latest --size Standard_B1ms --admin-username azureadmin --ssh-key-value ~/.ssh/id_rsa.pub --vnet-name vnet-cashflow --subnet snet-agent --nsg nsg-agent --os-disk-size-gb 32
az postgres flexible-server create --resource-group rg-cashflow-agent --name psql-cashflow-agent --location eastus2 --admin-user pgadmin --admin-password '<STRONG_PASSWORD>' --sku-name Standard_B1ms --tier Burstable --storage-size 32 --version 15
az postgres flexible-server firewall-rule create --resource-group rg-cashflow-agent --name psql-cashflow-agent --rule-name AllowAgentVM --start-ip-address <VM_PRIVATE_IP> --end-ip-address <VM_PRIVATE_IP>

Step 2: Configure Production Server Base Environment

SSH into the production VM and install Docker, Docker Compose, and essential system packages. Configure automatic security updates, UFW firewall, and fail2ban for SSH protection. Set the timezone and configure log rotation.

ssh azureadmin@<PROD_VM_PUBLIC_IP>
sudo apt update && sudo apt upgrade -y
sudo apt install -y ca-certificates curl gnupg lsb-release ufw fail2ban unattended-upgrades
sudo install -m 0755 -d /etc/apt/keyrings
curl -fsSL https://download.docker.com/linux/ubuntu/gpg | sudo gpg --dearmor -o /etc/apt/keyrings/docker.gpg
echo "deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.gpg] https://download.docker.com/linux/ubuntu $(lsb_release -cs) stable" | sudo tee /etc/apt/sources.list.d/docker.list > /dev/null
sudo apt update && sudo apt install -y docker-ce docker-ce-cli containerd.io docker-buildx-plugin docker-compose-plugin
sudo usermod -aG docker azureadmin
sudo systemctl enable docker
sudo ufw default deny incoming
sudo ufw default allow outgoing
sudo ufw allow from <MSP_IP_RANGE> to any port 22 proto tcp
sudo ufw allow 443/tcp
sudo ufw enable
sudo timedatectl set-timezone America/New_York
sudo dpkg-reconfigure -plow unattended-upgrades

Step 3: Deploy n8n with Docker Compose

Create the n8n deployment directory, Docker Compose configuration, and Nginx reverse proxy with Let's Encrypt SSL. n8n will be the central orchestration platform for all AI agent workflows. Configure it to use the Azure PostgreSQL database for workflow storage and enable webhook functionality.

mkdir -p /opt/cashflow-agent/n8n && cd /opt/cashflow-agent
cat > docker-compose.yml << 'EOF'
version: '3.8'
services:
  n8n:
    image: n8nio/n8n:latest
    container_name: n8n
    restart: unless-stopped
    ports:
      - '5678:5678'
    environment:
      - N8N_HOST=n8n.yourdomain.com
      - N8N_PORT=5678
      - N8N_PROTOCOL=https
      - WEBHOOK_URL=https://n8n.yourdomain.com/
      - N8N_ENCRYPTION_KEY=${N8N_ENCRYPTION_KEY}
      - DB_TYPE=postgresdb
      - DB_POSTGRESDB_HOST=${POSTGRES_HOST}
      - DB_POSTGRESDB_PORT=5432
      - DB_POSTGRESDB_DATABASE=n8n
      - DB_POSTGRESDB_USER=${POSTGRES_USER}
      - DB_POSTGRESDB_PASSWORD=${POSTGRES_PASSWORD}
      - DB_POSTGRESDB_SSL_REJECT_UNAUTHORIZED=false
      - N8N_BASIC_AUTH_ACTIVE=true
      - N8N_BASIC_AUTH_USER=${N8N_ADMIN_USER}
      - N8N_BASIC_AUTH_PASSWORD=${N8N_ADMIN_PASSWORD}
      - EXECUTIONS_DATA_PRUNE=true
      - EXECUTIONS_DATA_MAX_AGE=168
      - GENERIC_TIMEZONE=America/New_York
    volumes:
      - n8n_data:/home/node/.n8n
  
  nginx:
    image: nginx:alpine
    container_name: nginx-proxy
    restart: unless-stopped
    ports:
      - '443:443'
      - '80:80'
    volumes:
      - ./nginx/conf.d:/etc/nginx/conf.d
      - ./nginx/certbot/conf:/etc/letsencrypt
      - ./nginx/certbot/www:/var/www/certbot
    depends_on:
      - n8n

  certbot:
    image: certbot/certbot
    container_name: certbot
    volumes:
      - ./nginx/certbot/conf:/etc/letsencrypt
      - ./nginx/certbot/www:/var/www/certbot
    entrypoint: "/bin/sh -c 'trap exit TERM; while :; do certbot renew; sleep 12h & wait $${!}; done;'"

volumes:
  n8n_data:
EOF
cat > .env << 'EOF'
N8N_ENCRYPTION_KEY=<GENERATE_WITH_openssl_rand_-hex_32>
POSTGRES_HOST=psql-cashflow-agent.postgres.database.azure.com
POSTGRES_USER=pgadmin
POSTGRES_PASSWORD=<YOUR_POSTGRES_PASSWORD>
N8N_ADMIN_USER=admin
N8N_ADMIN_PASSWORD=<STRONG_ADMIN_PASSWORD>
EOF
chmod 600 .env
mkdir -p nginx/conf.d nginx/certbot/conf nginx/certbot/www
cat > nginx/conf.d/n8n.conf << 'EOF'
server {
    listen 80;
    server_name n8n.yourdomain.com;
    location /.well-known/acme-challenge/ {
        root /var/www/certbot;
    }
    location / {
        return 301 https://$host$request_uri;
    }
}
server {
    listen 443 ssl;
    server_name n8n.yourdomain.com;
    ssl_certificate /etc/letsencrypt/live/n8n.yourdomain.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/n8n.yourdomain.com/privkey.pem;
    location / {
        proxy_pass http://n8n:5678;
        proxy_http_version 1.1;
        proxy_set_header Upgrade $http_upgrade;
        proxy_set_header Connection 'upgrade';
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        chunked_transfer_encoding off;
        proxy_buffering off;
        proxy_cache off;
    }
}
EOF
# First start nginx without SSL to get certificate
# Temporarily modify nginx conf to serve only port 80
docker compose up -d nginx
docker compose run --rm certbot certonly --webroot -w /var/www/certbot -d n8n.yourdomain.com --email admin@yourdomain.com --agree-tos --no-eff-email
# Restore full nginx conf with SSL and start all services
docker compose up -d

Step 4: Create PostgreSQL Database Schema for Agent Data

Connect to the Azure PostgreSQL instance and create the application database, tables for transaction caching, alert history, client configuration, and audit logging. These tables support the AI agent's data pipeline and provide the compliance audit trail required by IRS Publication 4557.

psql "host=psql-cashflow-agent.postgres.database.azure.com port=5432 dbname=postgres user=pgadmin sslmode=require"
CREATE DATABASE cashflow_agent;
\c cashflow_agent
CREATE TABLE clients (
  id SERIAL PRIMARY KEY,
  name VARCHAR(255) NOT NULL,
  platform VARCHAR(20) NOT NULL CHECK (platform IN ('qbo', 'xero')),
  platform_entity_id VARCHAR(255) NOT NULL,
  oauth_access_token TEXT,
  oauth_refresh_token TEXT,
  oauth_token_expires_at TIMESTAMPTZ,
  min_cash_threshold NUMERIC(12,2) DEFAULT 5000.00,
  large_txn_threshold NUMERIC(12,2) DEFAULT 5000.00,
  alert_channel VARCHAR(50) DEFAULT 'slack',
  alert_destination TEXT,
  is_active BOOLEAN DEFAULT TRUE,
  created_at TIMESTAMPTZ DEFAULT NOW(),
  updated_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE TABLE transactions (
  id SERIAL PRIMARY KEY,
  client_id INTEGER REFERENCES clients(id),
  platform_txn_id VARCHAR(255) NOT NULL,
  txn_date DATE NOT NULL,
  amount NUMERIC(12,2) NOT NULL,
  description TEXT,
  category VARCHAR(255),
  ai_category VARCHAR(255),
  ai_confidence NUMERIC(4,3),
  is_anomaly BOOLEAN DEFAULT FALSE,
  anomaly_reason TEXT,
  raw_json JSONB,
  processed_at TIMESTAMPTZ DEFAULT NOW(),
  UNIQUE(client_id, platform_txn_id)
);
CREATE TABLE alerts (
  id SERIAL PRIMARY KEY,
  client_id INTEGER REFERENCES clients(id),
  alert_type VARCHAR(50) NOT NULL,
  severity VARCHAR(20) NOT NULL CHECK (severity IN ('info', 'warning', 'critical')),
  title VARCHAR(500) NOT NULL,
  body TEXT NOT NULL,
  ai_generated BOOLEAN DEFAULT TRUE,
  delivered BOOLEAN DEFAULT FALSE,
  delivered_at TIMESTAMPTZ,
  acknowledged BOOLEAN DEFAULT FALSE,
  acknowledged_by VARCHAR(255),
  created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE TABLE audit_log (
  id SERIAL PRIMARY KEY,
  event_type VARCHAR(100) NOT NULL,
  client_id INTEGER REFERENCES clients(id),
  details JSONB,
  created_at TIMESTAMPTZ DEFAULT NOW()
);
CREATE TABLE cash_snapshots (
  id SERIAL PRIMARY KEY,
  client_id INTEGER REFERENCES clients(id),
  snapshot_date DATE NOT NULL,
  total_bank_balance NUMERIC(14,2),
  accounts_receivable NUMERIC(14,2),
  accounts_payable NUMERIC(14,2),
  net_cash_position NUMERIC(14,2),
  days_cash_on_hand INTEGER,
  burn_rate_daily NUMERIC(12,2),
  created_at TIMESTAMPTZ DEFAULT NOW(),
  UNIQUE(client_id, snapshot_date)
);
CREATE INDEX idx_transactions_client_date ON transactions(client_id, txn_date DESC);
CREATE INDEX idx_alerts_client_created ON alerts(client_id, created_at DESC);
CREATE INDEX idx_snapshots_client_date ON cash_snapshots(client_id, snapshot_date DESC);

Step 5: Register and Configure QuickBooks Online API Application

Register a new application in the Intuit Developer portal to obtain OAuth 2.0 credentials. Configure the app with the correct scopes and redirect URIs for n8n integration. This enables programmatic access to bank feed transactions, account balances, invoices, and bills.

Step 6: Register and Configure Xero API Application

Register a new application in the Xero Developer portal for OAuth 2.0 access. Configure scopes for bank transactions, accounts, invoices, and contacts. Set up webhook subscriptions for real-time bank feed notifications.

Step 7: Configure n8n Credentials and Core Settings

Log into the n8n web interface and configure all required credentials: QBO OAuth2, Xero OAuth2, OpenAI API key, PostgreSQL database connection, and Slack/Teams webhook. These credentials are encrypted by n8n using the N8N_ENCRYPTION_KEY.

Step 8: Deploy Float and Connect Client Entities (Phase 1)

Set up Float as the baseline cash flow forecasting and threshold alerting tool. Connect each client's QBO or Xero entity to Float, configure cash threshold alerts, and verify initial data sync. This provides immediate value while the custom AI agent is being developed.

Step 9: Deploy Fathom for Portfolio Monitoring

Set up Fathom for portfolio-wide financial KPI tracking across all client entities. This complements Float by providing ratio analysis, trend detection, and automated reporting. Configure KPI alerts for cash-related metrics across the portfolio.

Step 10: Import and Activate n8n Agent Workflows

Import the custom AI agent workflows into n8n. These workflows implement the core autonomous monitoring pipeline: bank feed polling, transaction analysis, anomaly detection, cash flow snapshot generation, and alert delivery. Each workflow is described in detail in the Custom AI Components section of this guide.

Step 11: Configure Alert Channels and Test End-to-End

Set up the Slack workspace (or Teams/email) alert channels, configure the alert dispatcher workflow with channel mappings, and run a full end-to-end test by manually inserting a test transaction that triggers an alert.

Step 12: Configure Backup and Disaster Recovery

Set up automated backups for the PostgreSQL database, n8n workflow configurations, and Docker volumes. Configure backup verification and offsite storage to the Synology NAS or Azure Blob Storage.

# Create backup script
cat > /opt/cashflow-agent/backup.sh << 'SCRIPT'
#!/bin/bash
set -euo pipefail
BACKUP_DIR=/opt/cashflow-agent/backups/$(date +%Y%m%d)
mkdir -p $BACKUP_DIR

# Database backup
PGPASSWORD='<POSTGRES_PASSWORD>' pg_dump \
  -h psql-cashflow-agent.postgres.database.azure.com \
  -U pgadmin \
  -d cashflow_agent \
  -F c \
  -f $BACKUP_DIR/cashflow_agent_$(date +%H%M).dump

# n8n workflow export
docker exec n8n n8n export:workflow --all --output=$BACKUP_DIR/workflows.json 2>/dev/null || true

# Docker volume backup
docker run --rm -v cashflow-agent_n8n_data:/data -v $BACKUP_DIR:/backup alpine \
  tar czf /backup/n8n_data.tar.gz /data

# Configuration backup
cp /opt/cashflow-agent/docker-compose.yml $BACKUP_DIR/
cp /opt/cashflow-agent/.env $BACKUP_DIR/

# Encrypt backups
tar czf $BACKUP_DIR.tar.gz $BACKUP_DIR
gpg --symmetric --cipher-algo AES256 --passphrase-file /opt/cashflow-agent/.backup-key $BACKUP_DIR.tar.gz
rm -rf $BACKUP_DIR $BACKUP_DIR.tar.gz

# Upload to Azure Blob (or rsync to NAS)
az storage blob upload \
  --account-name stcashflowbackup \
  --container-name backups \
  --name $(date +%Y%m%d).tar.gz.gpg \
  --file $BACKUP_DIR.tar.gz.gpg \
  --auth-mode login 2>/dev/null || true

# Cleanup: keep last 30 days locally
find /opt/cashflow-agent/backups/ -name '*.gpg' -mtime +30 -delete

echo "Backup completed: $(date)"
SCRIPT
chmod +x /opt/cashflow-agent/backup.sh
# Generate backup encryption key
openssl rand -hex 32 > /opt/cashflow-agent/.backup-key
chmod 600 /opt/cashflow-agent/.backup-key
# Schedule nightly backups at 2 AM
echo '0 2 * * * /opt/cashflow-agent/backup.sh >> /var/log/cashflow-backup.log 2>&1' | crontab -
# For Synology NAS alternative, replace az storage blob upload with:
# rsync -avz --progress $BACKUP_DIR.tar.gz.gpg admin@<NAS_IP>:/volume1/backups/cashflow-agent/

Step 13: Deploy UptimeRobot Monitoring

Configure UptimeRobot to monitor all critical endpoints: n8n web interface, webhook receivers, database connectivity, and API health checks. Set up escalation alerts to MSP on-call personnel.

Step 14: Security Hardening and Compliance Documentation

Implement final security hardening measures, generate compliance documentation, and update the client's WISP to reflect the new AI monitoring system. This step ensures the deployment meets IRS Publication 4557 and FTC Safeguards Rule requirements.

Step 15: Roll Out to Full Client Portfolio

After successful testing with 1-3 pilot clients, systematically onboard the remaining client entities. Use a batch onboarding process to efficiently connect QBO/Xero entities, configure thresholds, and validate data flow for each client.

Custom AI Components

Bank Feed Poller Workflow

Type: workflow

An n8n workflow that runs on a 15-minute schedule (configurable per client). For each active client entity, it queries the QBO or Xero API for new bank transactions since the last poll timestamp. New transactions are inserted into the PostgreSQL transactions table and passed to the Transaction Analyzer workflow via an n8n trigger. Handles OAuth token refresh, rate limiting, pagination, and error logging.

Implementation:

n8n Workflow Specification (JSON-importable):

**Trigger:** Schedule Trigger — Every 15 minutes

**Node 1: Get Active Clients**
Type: Postgres
Operation: Execute Query
Query: SELECT id, name, platform, platform_entity_id, oauth_access_token, oauth_refresh_token, oauth_token_expires_at, min_cash_threshold, large_txn_threshold FROM clients WHERE is_active = TRUE;

**Node 2: Split Into Batches**
Type: SplitInBatches
Batch Size: 5 (process 5 clients concurrently to respect API rate limits)

**Node 3: Switch on Platform**
Type: Switch
Condition: {{$json.platform}} equals 'qbo' → QBO Branch, equals 'xero' → Xero Branch

**Node 4a: QBO — Fetch Bank Transactions**
Type: HTTP Request
Method: GET
URL: https://quickbooks.api.intuit.com/v3/company/{{$json.platform_entity_id}}/query?query=SELECT * FROM Purchase WHERE MetaData.LastUpdatedTime > '{{$now.minus(16, 'minutes').toISO()}}' ORDERBY MetaData.LastUpdatedTime&minorversion=73
Headers:
  Authorization: Bearer {{$json.oauth_access_token}}
  Accept: application/json
Authentication: None (token in header)
Retry on Fail: true, Max Retries: 3, Wait Between: 5000ms
On Error: Continue (errors handled in error branch)

Also fetch BankDeposit: SELECT * FROM Deposit WHERE MetaData.LastUpdatedTime > '{{$now.minus(16, 'minutes').toISO()}}'

**Node 4b: Xero — Fetch Bank Transactions**
Type: HTTP Request
Method: GET
URL: https://api.xero.com/api.xro/2.0/BankTransactions?where=UpdatedDateUTC>=DateTime({{$now.minus(16, 'minutes').format('yyyy,MM,dd,HH,mm,ss')}})&page=1
Headers:
  Authorization: Bearer {{$json.oauth_access_token}}
  Xero-Tenant-Id: {{$json.platform_entity_id}}
  Accept: application/json

**Node 5: Merge & Normalize Transactions**
Type: Code (JavaScript)
Code:
const items = $input.all();
const normalized = [];
for (const item of items) {
  const data = item.json;
  if (data.QueryResponse && data.QueryResponse.Purchase) {
    for (const txn of data.QueryResponse.Purchase) {
      normalized.push({
        client_id: data._client_id,
        platform_txn_id: txn.Id,
        txn_date: txn.TxnDate,
        amount: -Math.abs(txn.TotalAmt),
        description: txn.Line?.[0]?.Description || txn.EntityRef?.name || 'No description',
        category: txn.Line?.[0]?.AccountBasedExpenseLineDetail?.AccountRef?.name || 'Uncategorized',
        raw_json: txn
      });
    }
  }
  if (data.QueryResponse && data.QueryResponse.Deposit) {
    for (const txn of data.QueryResponse.Deposit) {
      normalized.push({
        client_id: data._client_id,
        platform_txn_id: txn.Id,
        txn_date: txn.TxnDate,
        amount: Math.abs(txn.TotalAmt),
        description: txn.Line?.[0]?.Description || 'Deposit',
        category: txn.Line?.[0]?.DepositLineDetail?.AccountRef?.name || 'Income',
        raw_json: txn
      });
    }
  }
  if (data.BankTransactions) {
    for (const txn of data.BankTransactions) {
      normalized.push({
        client_id: data._client_id,
        platform_txn_id: txn.BankTransactionID,
        txn_date: txn.Date,
        amount: txn.Type === 'RECEIVE' ? Math.abs(txn.Total) : -Math.abs(txn.Total),
        description: txn.Contact?.Name || txn.LineItems?.[0]?.Description || 'No description',
        category: txn.LineItems?.[0]?.AccountCode || 'Uncategorized',
        raw_json: txn
      });
    }
  }
}
return normalized.map(n => ({ json: n }));

**Node 6: Upsert Transactions**
Type: Postgres
Operation: Execute Query
Query (per item): INSERT INTO transactions (client_id, platform_txn_id, txn_date, amount, description, category, raw_json) VALUES ($1, $2, $3, $4, $5, $6, $7) ON CONFLICT (client_id, platform_txn_id) DO UPDATE SET amount = EXCLUDED.amount, description = EXCLUDED.description, category = EXCLUDED.category, raw_json = EXCLUDED.raw_json, processed_at = NOW() RETURNING id, client_id, platform_txn_id, amount, description;

**Node 7: Trigger Transaction Analyzer**
Type: Execute Workflow
Workflow: 02-transaction-analyzer
Pass data: All new/updated transaction IDs

**Node 8: Audit Log**
Type: Postgres
Operation: Insert
Table: audit_log
Values: event_type='bank_feed_poll', client_id={{$json.client_id}}, details={"transactions_found": count, "timestamp": now}

**Error Branch:**
Node E1: Check if error is 401 (token expired) → Trigger token refresh workflow
Node E2: Log error to audit_log with full error details
Node E3: If 3+ consecutive failures for a client, send alert to #cashflow-agent-errors

Transaction Analyzer Agent

Type: agent

An AI agent that analyzes each new bank transaction using GPT-5.4 mini for classification and anomaly scoring. For each transaction, it determines: (1) whether the category assigned by QBO/Xero is correct, (2) whether the transaction amount is anomalous based on historical patterns for this client, (3) whether the transaction represents a potential cash flow concern (e.g., unexpected large payment, duplicate transaction, unusual vendor). Transactions scoring above the anomaly threshold are forwarded to the Anomaly Detector for deeper analysis with GPT-5.4.

Implementation:

Transaction to Analyze

• Date: ${txn.txn_date}
• Amount: $${txn.amount}
• Description: ${txn.description}
• Current Category: ${txn.category}

Client Historical Context (Last 90 Days)

• Average transaction amount: $${stats.avg_txn_amount?.toFixed(2) || 'N/A'}
• Std deviation: $${stats.stddev_txn_amount?.toFixed(2) || 'N/A'}
• Max transaction: $${stats.max_txn_amount?.toFixed(2) || 'N/A'}
• Total transactions: ${stats.total_txns || 0}
• Average expense: $${stats.avg_expense?.toFixed(2) || 'N/A'}
• Average income: $${stats.avg_income?.toFixed(2) || 'N/A'}

Recent Transactions (Last 7 Days)

${recentTxns.map(t => ${t.txn_date} | $${t.amount} | ${t.description}).join('\n')}

Your Task

Analyze this transaction and respond with ONLY a JSON object (no markdown, no explanation):

{
  "suggested_category": "<correct category name>",
  "category_confidence": <0.0-1.0>,
  "is_anomaly": <true/false>,
  "anomaly_score": <0.0-1.0 where 1.0 is most anomalous>,
  "anomaly_reasons": ["<reason1>", "<reason2>"],
  "is_duplicate_suspect": <true/false>,
  "duplicate_of": "<platform_txn_id or null>",
  "cash_flow_impact": "<positive/negative/neutral>",
  "concern_level": "<none/low/medium/high/critical>",
  "brief_note": "<one-sentence plain-English summary for the bookkeeper>"
}

Anomaly detection rules:

`;

return [{ json: { prompt, txn, stats } }];

Node 4: Call GPT-5.4 mini

• Type: OpenAI
• Resource: Chat Completion
• Model: gpt-5.4-mini
• Messages: [{role: 'user', content: '{{$json.prompt}}'}]
• Temperature: 0.1
• Max Tokens: 500
• Response Format: json_object

Node 5: Parse and Store Results

• Type: Code (JavaScript)

const analysis = JSON.parse($input.first().json.message.content);
const txn = $input.first().json.txn;

return [{
  json: {
    transaction_id: txn.id,
    client_id: txn.client_id,
    ai_category: analysis.suggested_category,
    ai_confidence: analysis.category_confidence,
    is_anomaly: analysis.anomaly_score >= 0.5 || analysis.is_duplicate_suspect,
    anomaly_reason: analysis.anomaly_reasons?.join('; ') || null,
    anomaly_score: analysis.anomaly_score,
    concern_level: analysis.concern_level,
    brief_note: analysis.brief_note,
    full_analysis: analysis
  }
}];

Node 6: Update Transaction Record

• Type: Postgres

UPDATE transactions SET 
  ai_category = $1, 
  ai_confidence = $2, 
  is_anomaly = $3, 
  anomaly_reason = $4 
WHERE id = $5;

Node 7: Route Anomalies

• Type: IF
• Condition: anomaly_score >= 0.5 OR concern_level IN ('high', 'critical')
• True → Trigger Anomaly Detector (03-anomaly-detector)
• False → End (normal transaction, no further action)

Node 8: Audit Log

• Type: Postgres
• Insert into audit_log: event_type='transaction_analyzed', details={analysis summary}

Anomaly Detector Deep Analysis Agent

• Type: agent

A second-tier AI agent using GPT-5.4 (the more capable model) that performs deep analysis on transactions flagged by the Transaction Analyzer. It receives the flagged transaction along with broader client financial context (30-day transaction history, current cash position, upcoming payables) and generates a detailed anomaly report with recommended actions. This agent only runs on flagged transactions to control API costs.

Implementation:

n8n Workflow Specification:

Trigger: Called by Transaction Analyzer when anomaly_score >= 0.5

Node 1: Receive Flagged Transaction

• Type: Execute Workflow Trigger

Node 2: Fetch Extended Context

• Type: Postgres (multiple queries)

Query 1 - 30-day transaction history:
SELECT txn_date, amount, description, category, ai_category, is_anomaly
FROM transactions
WHERE client_id = $1 AND txn_date > CURRENT_DATE - INTERVAL '30 days'
ORDER BY txn_date DESC;
Query 2 - Latest cash snapshot:
SELECT * FROM cash_snapshots
WHERE client_id = $1
ORDER BY snapshot_date DESC LIMIT 1;
Query 3 - Client thresholds:
SELECT min_cash_threshold, large_txn_threshold FROM clients WHERE id = $1;

Node 3: Build Deep Analysis Prompt

• Type: Code (JavaScript)

const flagged = $input.first().json;
const history = $input.item(1).json.rows;
const snapshot = $input.item(2).json.rows[0];
const thresholds = $input.item(3).json.rows[0];

const prompt = `You are a senior forensic accountant AI agent. A transaction has been flagged as potentially anomalous. Perform a deep analysis and generate an actionable alert.

Flagged Transaction

• Date: ${flagged.txn_date}
• Amount: $${flagged.amount}
• Description: ${flagged.description}
• Current Category: ${flagged.category}
• AI Suggested Category: ${flagged.ai_category}
• Initial Anomaly Score: ${flagged.anomaly_score}
• Initial Flags: ${flagged.anomaly_reason}

Current Cash Position

• Total Bank Balance: $${snapshot?.total_bank_balance || 'Unknown'}
• Accounts Receivable: $${snapshot?.accounts_receivable || 'Unknown'}
• Accounts Payable: $${snapshot?.accounts_payable || 'Unknown'}
• Net Cash Position: $${snapshot?.net_cash_position || 'Unknown'}
• Days Cash on Hand: ${snapshot?.days_cash_on_hand || 'Unknown'}
• Daily Burn Rate: $${snapshot?.burn_rate_daily || 'Unknown'}

Alert Thresholds

• Minimum Cash Balance: $${thresholds.min_cash_threshold}
• Large Transaction Threshold: $${thresholds.large_txn_threshold}

30-Day Transaction History

${history.map(t => ${t.txn_date} | $${t.amount} | ${t.description} | ${t.category} ${t.is_anomaly ? '⚠️' : ''}).join('\n')}

Your Analysis Tasks

1. Confirm or revise the anomaly classification 2. Assess the cash flow impact of this transaction 3. Determine if this transaction, combined with current cash position, creates a cash flow risk 4. Check for patterns (is this part of a series of unusual transactions?) 5. Generate a clear, actionable alert for the bookkeeper Respond with ONLY a JSON object: { "confirmed_anomaly": <true/false>, "revised_anomaly_score": <0.0-1.0>, "alert_severity": "<info|warning|critical>", "alert_title":...

Portfolio Cash Positions (Today)

• ${snapshots.map(s => - ${s.name}: Bank $${Number(s.total_bank_balance).toLocaleString()} | AR $${Number(s.accounts_receivable).toLocaleString()} | AP $${Number(s.accounts_payable).toLocaleString()} | Net $${Number(s.net_cash_position).toLocaleString()} | ${s.days_cash_on_hand} days cash | Burn $${Number(s.burn_rate_daily).toLocaleString()}/day).join('\n')}

Alerts in Last 24 Hours (${alerts.length} total)

• ${alerts.length > 0 ? alerts.map(a => - [${a.severity.toUpperCase()}] ${a.client_name}: ${a.title}).join('\n') : 'No alerts triggered.'}

Anomalous Transactions (${anomalies.length} flagged)

• ${anomalies.length > 0 ? anomalies.map(a => - ${a.client_name}: $${a.amount} — ${a.description} (${a.anomaly_reason})).join('\n') : 'No anomalies detected.'}

Instructions

Generate a morning briefing with these sections:

Keep the total briefing under 400 words. Use plain English, avoid jargon. Address the team directly.`;

return [{ json: { prompt } }];

Node 5: Call GPT-5.4

• Type: OpenAI
• Model: gpt-5.4
• Messages: [{role: 'system', content: 'You are a concise, professional financial advisor AI. Generate clear, actionable briefings.'}, {role: 'user', content: '{{$json.prompt}}'}]
• Temperature: 0.3
• Max Tokens: 800

Node 6: Format and Send to Slack

• Type: Slack
• Channel: #cashflow-alerts-info
• Message: ☀️ *Morning Cash Flow Briefing — {{$now.format('MMMM d, yyyy')}}* {{$json.message.content}} --- _Generated by CashFlow Monitor Agent at {{$now.format('h:mm a')}}_

Node 7: Also send via email (optional)

• Type: Send Email
• To: accounting-team@clientfirm.com
• Subject: Daily Cash Flow Briefing — {{$now.format('MMM d')}}
• Body: HTML-formatted version of the digest

OAuth Token Refresher

• Type: workflow

A utility workflow that runs every 45 minutes to proactively refresh OAuth tokens for QBO and Xero before they expire. QBO tokens expire every 60 minutes; Xero tokens expire every 30 minutes. This ensures uninterrupted API access for all agent workflows.

Implementation:

n8n Workflow Specification:

**Trigger:** Schedule — Every 45 minutes

**Node 1: Get Clients Needing Token Refresh**
Type: Postgres
SELECT id, name, platform, platform_entity_id, oauth_refresh_token, oauth_token_expires_at
FROM clients
WHERE is_active = TRUE
  AND oauth_token_expires_at < NOW() + INTERVAL '20 minutes';

**Node 2: For Each Client — Refresh Token**

For QBO:
Type: HTTP Request
Method: POST
URL: https://oauth.platform.intuit.com/oauth2/v1/tokens/bearer
Body (form-urlencoded):
  grant_type=refresh_token
  refresh_token={{$json.oauth_refresh_token}}
Headers:
  Authorization: Basic {{base64(client_id + ':' + client_secret)}}
  Content-Type: application/x-www-form-urlencoded

For Xero:
Type: HTTP Request
Method: POST
URL: https://identity.xero.com/connect/token
Body (form-urlencoded):
  grant_type=refresh_token
  refresh_token={{$json.oauth_refresh_token}}
  client_id={{xero_client_id}}
  client_secret={{xero_client_secret}}

**Node 3: Update Tokens in Database**
Type: Postgres
UPDATE clients SET
  oauth_access_token = $1,
  oauth_refresh_token = $2,
  oauth_token_expires_at = NOW() + INTERVAL '$3 seconds',
  updated_at = NOW()
WHERE id = $4;

**Node 4: Error Handling**
If refresh fails (e.g., refresh token expired after 100 days for QBO):
- Log error to audit_log
- Send critical alert to #cashflow-agent-errors
- Set client.is_active = FALSE
- Alert MSP: 'Re-authorization required for {client_name}'

**Node 5: Audit Log**
Log all token refresh events for compliance trail.

Cash Flow Analysis System Prompt

• Type: prompt

The master system prompt used across all LLM calls in the agent pipeline. Establishes the AI's persona, capabilities, constraints, and output formatting requirements. This prompt is referenced by the Transaction Analyzer and Anomaly Detector agents.

Implementation:

Your Role

You work for an accounting firm that manages bookkeeping for multiple business clients. Your job is to continuously monitor bank transactions and cash positions, detect anomalies, and surface actionable alerts before cash flow problems become crises.

Your Capabilities

Your Constraints

• NEVER fabricate or hallucinate financial data. If you lack data to make a calculation, explicitly state 'Insufficient data' rather than estimating.
• NEVER provide tax advice or legal opinions.
• ALWAYS express amounts in USD with appropriate precision (dollars and cents for transactions, whole dollars for summaries).
• ALWAYS classify alert severity conservatively — it's better to over-alert than miss a real problem.
• Your anomaly detection should have a FALSE POSITIVE rate target of <15%. Err toward flagging suspicious items.
• NEVER include raw API data, OAuth tokens, or system internals in alert messages.
• Alerts must be written for a bookkeeper with 2-5 years of experience — avoid AI jargon.
• When recommending actions, be specific: 'Contact [vendor name] about the [varies],XXX payment on [date]' not 'Investigate the transaction.'

Severity Definitions

• CRITICAL: Immediate cash flow risk. Cash below threshold, projected shortfall within 7 days, or transaction suggesting fraud/error >$10,000.
• WARNING: Emerging concern. Projected shortfall within 14-30 days, unusual transaction >$5,000, spending rate increase >25%, or potential duplicate payment.
• INFO: Noteworthy but not urgent. New vendor payments, category corrections, positive cash flow trends, or monthly pattern observations.

Output Format

Always respond with valid JSON matching the schema requested in the user prompt. Do not include markdown formatting, code fences, or explanatory text outside the JSON object.

Testing & Validation

• SMOKE TEST — n8n Accessibility: Navigate to https://n8n.yourdomain.com in a browser. Verify the login page loads over HTTPS with a valid SSL certificate. Log in with admin credentials and confirm the dashboard displays all imported workflows.
• API CONNECTIVITY — QuickBooks Online: In n8n, create a temporary workflow with an HTTP Request node querying GET https://quickbooks.api.intuit.com/v3/company/<REALM_ID>/companyinfo/<REALM_ID>?minorversion=73 using the QBO OAuth2 credential. Execute and verify a 200 response with valid company info JSON.
• API CONNECTIVITY — Xero: In n8n, create a temporary workflow with an HTTP Request node querying GET https://api.xero.com/api.xro/2.0/Organisation using the Xero OAuth2 credential. Execute and verify a 200 response with valid organization data.
• API CONNECTIVITY — OpenAI: In n8n, use the OpenAI node to send a simple chat completion request: 'Respond with the word OK'. Verify a 200 response and the model returns 'OK'. Check that billing is active on the OpenAI dashboard.
• DATABASE — PostgreSQL Connection: Execute a test query from n8n's Postgres node: SELECT NOW(), version(); Verify it returns the current timestamp and PostgreSQL 15.x version string. Then verify all custom tables exist: SELECT table_name FROM information_schema.tables WHERE table_schema = 'public';
• BANK FEED POLLER — Single Client: Activate the Bank Feed Poller workflow. Manually trigger it or wait for the 15-minute schedule. Check the n8n Executions tab for a successful run. Then query: SELECT COUNT(*) FROM transactions WHERE client_id = <TEST_CLIENT_ID> AND processed_at > NOW() - INTERVAL '1 hour'; Verify transactions were inserted.
• TRANSACTION ANALYZER — Classification Accuracy: Insert 10 known test transactions covering various categories (payroll, rent, utilities, office supplies, client payment) into the test client's bank feed. Run the analyzer and verify: (1) ai_category is assigned for all 10, (2) ai_confidence > 0.7 for at least 8/10, (3) category suggestions are reasonable for the transaction descriptions.
• ANOMALY DETECTION — True Positive: Create a test bank transaction with an unusually large amount ($50,000 when average is $2,000). Run the pipeline and verify: (1) Transaction Analyzer flags it with anomaly_score >= 0.7, (2) Anomaly Detector generates a detailed analysis, (3) An alert record is created in the alerts table with severity 'warning' or 'critical'.
• ANOMALY DETECTION — Duplicate Detection: Create two bank transactions with the same amount ($1,547.32) and similar descriptions within 3 days. Run the pipeline and verify the second transaction is flagged with is_duplicate_suspect = true.
• ANOMALY DETECTION — False Positive Rate: Process 100 normal historical transactions through the analyzer. Count how many are flagged as anomalies. The false positive rate should be below 15% (fewer than 15 flagged). If higher, adjust the anomaly_score threshold in the Transaction Analyzer workflow.
• CASH SNAPSHOT — Daily Generation: Manually trigger the Cash Snapshot Generator for the test client. Verify: (1) A record is inserted in cash_snapshots with today's date, (2) total_bank_balance matches the value shown in QBO/Xero, (3) days_cash_on_hand is calculated correctly (bank balance / daily burn rate).
• THRESHOLD ALERT — Cash Below Minimum: Set the test client's min_cash_threshold to a value above their current bank balance. Trigger the snapshot generator. Verify a CRITICAL alert is generated with title containing 'below minimum threshold' and delivered to the correct Slack channel.
• THRESHOLD ALERT — Projected Breach: Set the test client's min_cash_threshold to a value that their current burn rate would breach in 10 days. Trigger the snapshot generator. Verify a WARNING alert about projected threshold breach within 14 days.
• ALERT DELIVERY — Slack: Verify alerts appear in the correct Slack channels: #cashflow-alerts-critical for warning/critical, #cashflow-alerts-info for info. Check that the Slack Block Kit formatting renders correctly with emoji, bold text, and action buttons.
• ALERT DELIVERY — Deduplication: Trigger the same alert condition twice within 4 hours. Verify that only one Slack message is sent (deduplication working).
• DAILY DIGEST — Generation and Delivery: Manually trigger the Daily Portfolio Digest Generator. Verify: (1) A well-formatted morning briefing appears in #cashflow-alerts-info, (2) The briefing includes all active clients, (3) Clients with low cash are highlighted, (4) The briefing is under 400 words and uses plain English.
• OAUTH TOKEN REFRESH — Automatic: Check the audit_log for recent token refresh events: SELECT * FROM audit_log WHERE event_type = 'token_refresh' ORDER BY created_at DESC LIMIT 10; Verify tokens are being refreshed every 45 minutes for clients with expiring tokens.
• BACKUP — Database Restore: Run the backup script manually: /opt/cashflow-agent/backup.sh. Then restore the backup to the staging database: gpg -d <backup>.tar.gz.gpg | tar xz && pg_restore -h <staging_host> -U pgadmin -d cashflow_agent_test <backup_file>.dump. Verify all tables and data are intact.
• MONITORING — UptimeRobot: Temporarily stop the n8n Docker container (docker stop n8n). Verify UptimeRobot sends a down alert within 5 minutes. Restart n8n (docker start n8n) and verify UptimeRobot sends an up notification.
• SECURITY — SSL and Encryption: Run SSL Labs test (ssllabs.com/ssltest) against n8n.yourdomain.com. Verify grade A or higher. Verify PostgreSQL connections require SSL: SELECT ssl FROM pg_stat_ssl WHERE pid = pg_backend_pid(); should return true.
• END-TO-END — Full Pipeline: Create a new bank transaction in the QBO/Xero sandbox. Time the full pipeline: transaction creation → polling detection → analysis → anomaly check → alert delivery. Target: under 20 minutes from transaction to alert. Document actual time for baseline.
• LOAD TEST — 50 Client Simulation: Configure all 50 client entities as active. Trigger a simultaneous bank feed poll for all clients. Monitor n8n execution queue, API rate limits (check for 429 errors), and VM resource utilization (CPU, RAM). Verify all polls complete within the 15-minute window without errors.

Client Handoff

Client Handoff Checklist

Training Session (2 hours, in-person or video)

Documentation to Deliver

• Operations Runbook (PDF): System architecture, all credentials (stored in shared password manager), alert severity definitions, response procedures, escalation contacts
• Client Configuration Spreadsheet: All monitored entities with thresholds, alert channels, and OAuth connection status
• WISP Amendment: Signed document updating the firm's Written Information Security Plan to include the AI monitoring system
• Vendor Contact Sheet: Support contacts for Float, Fathom, n8n (community), OpenAI, and Azure
• Monthly Report Template: Template for the MSP's monthly service report showing alert volume, false positive rate, system uptime, and recommendations

Success Criteria to Review Together

Ongoing Maintenance Responsibilities

Daily (Automated, MSP Monitors)

• Alert Pipeline Health: UptimeRobot monitors n8n uptime, webhook responsiveness, and API connectivity. MSP receives alerts on any downtime within 5 minutes. Automated n8n workflow checks database connectivity and OAuth token validity every 30 minutes.
• Execution Monitoring: Review n8n Executions tab daily (or configure error notifications) for failed workflow executions. Common failures: expired OAuth tokens (auto-refreshed, but refresh tokens expire after 100 days for QBO), API rate limits, temporary API outages.

Weekly (MSP Technician — 30 minutes)

• False Positive Review: Query alerts table for the week: SELECT severity, COUNT(*) FROM alerts WHERE created_at > NOW() - INTERVAL '7 days' GROUP BY severity; Review any alerts marked as acknowledged without action (likely false positives). If false positive rate exceeds 15%, schedule prompt tuning.
• System Resource Check: Verify Azure VM CPU/RAM utilization averages below 70%. Check PostgreSQL storage usage and vacuum status. Review Docker container logs for warnings.
• Backup Verification: Confirm nightly backups completed successfully: ls -la /opt/cashflow-agent/backups/ and check the most recent backup date.

Monthly (MSP Engineer — 2 hours)

• Prompt Tuning: Analyze the month's false positives and missed detections. Adjust anomaly_score thresholds, update the system prompt with new patterns, and test changes on the staging environment before promoting to production.
• Software Updates: Update n8n to the latest stable version (docker pull n8nio/n8n:latest && docker compose up -d). Review OpenAI model deprecation notices and update model references if needed. Apply Ubuntu security patches (automated via unattended-upgrades, but verify).
• Client Threshold Review: Generate a report of threshold breaches and near-misses. Recommend threshold adjustments to the accounting firm based on seasonal patterns.
• Monthly Service Report: Deliver to the client: alert volume by severity, false positive rate, system uptime percentage, notable detections, and recommendations for threshold changes.

Quarterly (MSP Engineer — 4 hours)

• Compliance Audit: Review WISP amendment for accuracy. Verify all vendor SOC 2 certifications are current. Test backup restoration to staging. Review access logs and remove any unused credentials.
• Performance Optimization: Analyze LLM API costs and optimize: are there transaction types that can be ruled out without LLM analysis? Can GPT-5.4 mini handle more of the anomaly detection load? Review database query performance and add indexes if needed.
• Client Onboarding/Offboarding: Add new client entities, remove departed clients, update OAuth connections as needed.
• Disaster Recovery Drill: Restore from backup to the staging VM and verify full pipeline functionality.

Annual

• Full Security Review: Rotate all API keys, OAuth app secrets, database passwords, and encryption keys. Review and update the WISP. Conduct a penetration test on the n8n instance.
• Architecture Review: Evaluate new tools and AI models. Consider upgrading from GPT-5.4 to newer models. Assess whether Intuit Assist or Xero's native AI has caught up to the custom agent's capabilities.
• Contract Review: Review managed service pricing against actual costs and effort. Adjust pricing for renewals.

SLA Commitments

• System Uptime: 99.5% monthly (allows ~3.6 hours downtime per month)
• Alert Delivery: Within 20 minutes of transaction detection during business hours
• Critical Issue Response: 4-hour MSP response time during business hours, next business day for non-critical
• Threshold Changes: Implemented within 1 business day of request
• New Client Onboarding: Completed within 5 business days of receiving OAuth authorization

Escalation Path