Supply Chain Exception Management Automation

Automatically detect, analyze, and resolve supply chain exceptions—delivery delays, invoice discrepancies, documentation gaps, temperature violations, and compliance issues. This guide shows you how to build an intelligent logistics exception management system that reduces manual coordination time by 82% while improving on-time delivery and carrier relationships.

Overview

Time Savings: 82% reduction in exception management time (40 hours/week → 7 hours/week) Complexity: Advanced Prerequisites: Access to TMS, carrier APIs, invoice systems, basic understanding of supply chain operations

Blocks & Tools You'll Use

Core Blocks

• Agent Block - For intelligent exception analysis, root cause detection, and resolution recommendations
• Webhook Block - To receive real-time shipment tracking updates from carrier APIs
• Function Block - For custom Python logic to calculate variances, validate rates, and score carrier performance
• Condition Block - To route exceptions based on severity, type, and resolution requirements
• Loop Block - To process batches of shipments and invoices systematically

API & Integration Tools

• HTTP Request Tool - To connect with TMS APIs, carrier tracking systems, and invoice platforms
• Google Sheets Tool - To maintain shipment tracking dashboards and performance analytics
• File Tool - To process shipping documents, invoices, and proof of delivery PDFs

Communication Tools

• Mail Tool - To send exception notifications to coordinators, carriers, and customers
• Slack Tool - For real-time alerts on critical exceptions
• Response Block - To deliver exception reports and resolution summaries

Data Storage

• Knowledge Base Tool - To store carrier contracts, rate tables, compliance requirements, and historical exception patterns

Workflow Overview

Data Integration (Webhook + HTTP Request) → Exception Detection (Function + Condition) → Root Cause Analysis (Agent + Knowledge Base) → Automated Resolution (Agent + Mail/Slack) → Performance Analytics (Function + Google Sheets)

Step 1: Unified Data Integration

Connect to TMS and Carrier APIs

Use the HTTP Request Tool and Webhook Block to integrate with your logistics systems.

TMS Integration Example (HTTP Request):

// Fetch all active shipments from TMS
const tmsApiUrl = 'https://your-tms.com/api/v1/shipments';
const headers = {
  'Authorization': `Bearer ${process.env.TMS_API_KEY}`,
  'Content-Type': 'application/json'
};

const params = {
  status: 'in_transit',
  date_from: new Date(Date.now() - 7 * 24 * 60 * 60 * 1000).toISOString()
};

// Returns list of active shipments with:
// - shipment_id, origin, destination, carrier, weight
// - expected_delivery_date, tracking_number, special_handling

Carrier Tracking Webhook Setup:

// Configure webhook endpoint in Klyntos to receive carrier tracking updates
// Carrier POSTs tracking events to: https://app.klyntos.com/webhooks/your-webhook-id

// Example webhook payload from carrier:
{
  "event_type": "tracking_update",
  "shipment_id": "SHP-12345",
  "tracking_number": "1Z999AA10123456784",
  "status": "in_transit",
  "current_location": {
    "city": "Chicago",
    "state": "IL",
    "timestamp": "2025-01-15T14:30:00Z"
  },
  "estimated_delivery": "2025-01-17T16:00:00Z",
  "exceptions": []
}

Unified Shipment View (Function Block):

def create_unified_shipment_view(tms_data, carrier_tracking, invoice_data):
    """
    Combine data from TMS, carrier tracking, and invoices into unified view.

    Returns comprehensive shipment record for exception monitoring.
    """
    unified_shipment = {
        'shipment_id': tms_data['shipment_id'],
        'tracking_number': tms_data['tracking_number'],

        # Origin and destination
        'origin': {
            'city': tms_data['origin_city'],
            'state': tms_data['origin_state'],
            'zip': tms_data['origin_zip']
        },
        'destination': {
            'city': tms_data['dest_city'],
            'state': tms_data['dest_state'],
            'zip': tms_data['dest_zip']
        },

        # Shipment details
        'carrier': tms_data['carrier_name'],
        'carrier_scac': tms_data['carrier_scac'],
        'service_level': tms_data['service_level'],  # e.g., "Ground", "2-Day", "Next Day"
        'weight_lbs': tms_data['weight'],
        'declared_value': tms_data['declared_value'],
        'special_handling': tms_data.get('special_handling', []),  # e.g., ["temperature_controlled", "fragile"]

        # Timing
        'ship_date': tms_data['ship_date'],
        'expected_delivery_date': tms_data['expected_delivery_date'],
        'actual_delivery_date': carrier_tracking.get('actual_delivery_date'),

        # Tracking status
        'current_status': carrier_tracking.get('status'),
        'current_location': carrier_tracking.get('current_location'),
        'last_tracking_update': carrier_tracking.get('last_update_timestamp'),
        'carrier_estimated_delivery': carrier_tracking.get('estimated_delivery'),

        # Invoice data
        'invoice_received': invoice_data is not None,
        'invoice_amount': invoice_data.get('total_amount') if invoice_data else None,
        'invoice_weight': invoice_data.get('weight') if invoice_data else None,
        'invoice_rate': invoice_data.get('rate_per_lb') if invoice_data else None,

        # Documentation
        'documents': {
            'bol': tms_data.get('bol_document_url'),
            'pod': carrier_tracking.get('proof_of_delivery_url'),
            'weight_ticket': None,  # populated later if received
            'customs_docs': None
        }
    }

    return unified_shipment

Step 2: Real-Time Exception Detection

Use Function Block + Condition Block to detect exceptions based on predefined thresholds.

Exception Detection Logic

from datetime import datetime, timedelta

def detect_shipment_exceptions(shipment, carrier_contract):
    """
    Detect all exceptions for a shipment.

    Returns list of exceptions with severity and type.
    """
    exceptions = []
    now = datetime.now()

    # 1. Delivery Delay Exception
    expected_delivery = datetime.fromisoformat(shipment['expected_delivery_date'])
    carrier_estimated = datetime.fromisoformat(shipment.get('carrier_estimated_delivery', shipment['expected_delivery_date']))

    if shipment['actual_delivery_date']:
        actual_delivery = datetime.fromisoformat(shipment['actual_delivery_date'])
        hours_late = (actual_delivery - expected_delivery).total_seconds() / 3600

        if hours_late > 24:
            exceptions.append({
                'type': 'delivery_delay',
                'severity': 'critical',
                'hours_late': round(hours_late, 1),
                'description': f"Shipment delivered {round(hours_late)} hours late",
                'expected': expected_delivery.isoformat(),
                'actual': actual_delivery.isoformat()
            })
        elif hours_late > 4:
            exceptions.append({
                'type': 'delivery_delay',
                'severity': 'warning',
                'hours_late': round(hours_late, 1),
                'description': f"Shipment delivered {round(hours_late)} hours late"
            })

    # Check for predicted delay (shipment in transit but carrier ETA moved)
    elif shipment['current_status'] == 'in_transit':
        predicted_delay_hours = (carrier_estimated - expected_delivery).total_seconds() / 3600

        if predicted_delay_hours > 12:
            exceptions.append({
                'type': 'predicted_delay',
                'severity': 'warning',
                'predicted_delay_hours': round(predicted_delay_hours, 1),
                'description': f"Carrier estimates delivery {round(predicted_delay_hours)} hours later than expected",
                'carrier_eta': carrier_estimated.isoformat()
            })

    # 2. Weight Discrepancy Exception
    if shipment['invoice_received'] and shipment['invoice_weight']:
        weight_diff_pct = abs(shipment['invoice_weight'] - shipment['weight_lbs']) / shipment['weight_lbs'] * 100

        if weight_diff_pct > 10:  # >10% discrepancy
            exceptions.append({
                'type': 'weight_discrepancy',
                'severity': 'warning',
                'tms_weight': shipment['weight_lbs'],
                'invoice_weight': shipment['invoice_weight'],
                'variance_pct': round(weight_diff_pct, 1),
                'description': f"Invoice weight differs from TMS by {round(weight_diff_pct)}%"
            })

    # 3. Invoice Rate Exception
    if shipment['invoice_received'] and carrier_contract:
        contracted_rate = get_contracted_rate(
            carrier_contract,
            shipment['origin'],
            shipment['destination'],
            shipment['weight_lbs']
        )

        if shipment['invoice_rate'] and contracted_rate:
            rate_diff_pct = abs(shipment['invoice_rate'] - contracted_rate) / contracted_rate * 100

            if rate_diff_pct > 5:
                exceptions.append({
                    'type': 'invoice_rate_anomaly',
                    'severity': 'warning',
                    'contracted_rate': contracted_rate,
                    'invoiced_rate': shipment['invoice_rate'],
                    'variance_pct': round(rate_diff_pct, 1),
                    'description': f"Invoice rate is {round(rate_diff_pct)}% different from contracted rate",
                    'potential_overcharge': round((shipment['invoice_rate'] - contracted_rate) * shipment['weight_lbs'], 2)
                })

    # 4. Temperature Violation Exception (if applicable)
    if 'temperature_controlled' in shipment.get('special_handling', []):
        temp_events = shipment.get('temperature_events', [])

        for event in temp_events:
            if event['temp_f'] < event['min_temp'] or event['temp_f'] > event['max_temp']:
                exceptions.append({
                    'type': 'temperature_violation',
                    'severity': 'critical',
                    'recorded_temp': event['temp_f'],
                    'acceptable_range': f"{event['min_temp']}-{event['max_temp']}°F",
                    'timestamp': event['timestamp'],
                    'description': f"Temperature violation: {event['temp_f']}°F (acceptable: {event['min_temp']}-{event['max_temp']}°F)"
                })

    # 5. Documentation Gap Exception
    required_docs = determine_required_documents(shipment)
    missing_docs = []

    for doc_type in required_docs:
        if not shipment['documents'].get(doc_type):
            missing_docs.append(doc_type)

    if missing_docs:
        exceptions.append({
            'type': 'documentation_gap',
            'severity': 'warning',
            'missing_documents': missing_docs,
            'description': f"Missing required documents: {', '.join(missing_docs)}"
        })

    # 6. Tracking Staleness Exception
    if shipment['current_status'] == 'in_transit':
        last_update = datetime.fromisoformat(shipment.get('last_tracking_update', now.isoformat()))
        hours_since_update = (now - last_update).total_seconds() / 3600

        if hours_since_update > 48:
            exceptions.append({
                'type': 'tracking_stale',
                'severity': 'warning',
                'hours_since_update': round(hours_since_update, 1),
                'description': f"No tracking update in {round(hours_since_update)} hours",
                'last_update': last_update.isoformat()
            })

    return exceptions

def get_contracted_rate(contract, origin, destination, weight_lbs):
    """
    Look up contracted rate from carrier contract based on lane and weight.
    """
    # Find matching lane in contract (origin zip -> destination zip)
    for lane in contract['lanes']:
        if (origin['zip'][:3] == lane['origin_zip_prefix'] and
            destination['zip'][:3] == lane['dest_zip_prefix']):

            # Find weight bracket
            for bracket in lane['rate_table']:
                if bracket['min_weight'] <= weight_lbs <= bracket['max_weight']:
                    return bracket['rate_per_lb']

    return None

def determine_required_documents(shipment):
    """
    Determine which documents are required for this shipment.
    """
    required = ['bol', 'pod']  # Always required

    # International shipments need customs docs
    if shipment['destination']['state'] in ['international', 'canada', 'mexico']:
        required.append('customs_docs')

    # High-value shipments need weight tickets
    if shipment['declared_value'] > 10000:
        required.append('weight_ticket')

    return required

Condition Block Routing

IF exception.severity == 'critical':
    → Route to immediate resolution workflow (Step 3)
    → Send Slack alert to logistics manager

ELIF exception.severity == 'warning':
    → Route to analysis workflow (Step 3)
    → Add to daily exception report

ELSE:
    → Log for trend analysis
    → No immediate action

Step 3: Intelligent Root Cause Analysis

Use Agent Block + Knowledge Base to analyze why exceptions occurred.

Agent Configuration for Root Cause Analysis

System Prompt:

You are a logistics exception analyst investigating the root cause of supply chain exceptions.

TASK:
Given a shipment exception, analyze possible root causes and recommend resolution steps.

INPUT:
- Exception details (type, severity, metrics)
- Shipment information (origin, destination, carrier, tracking history)
- Historical carrier performance data (from Knowledge Base)
- Carrier contract terms (from Knowledge Base)

ANALYSIS FRAMEWORK:

**For Delivery Delays:**
1. Check if delay is carrier-systemic (is this carrier consistently late on this lane?)
2. Check for external factors (weather, traffic, port congestion)
3. Check for mis-routed shipments (did carrier send it the wrong direction?)
4. Check for customs/inspection holds (international shipments)

**For Invoice Discrepancies:**
1. Verify weight: Does invoice weight match scale ticket or TMS estimate?
2. Check rate: Does invoice rate match contracted rate for this lane/weight bracket?
3. Check accessorials: Are there unexpected fuel surcharges, residential delivery fees, or dimensional weight charges?
4. Check contract compliance: Did carrier apply correct discount or base rate?

**For Documentation Gaps:**
1. Identify who is responsible for providing missing documentation (carrier, shipper, customs broker)
2. Check if documentation is overdue (expected by X date, now past due)
3. Determine compliance impact (does missing doc create regulatory exposure?)

OUTPUT FORMAT (JSON):
{
  "root_cause_analysis": {
    "primary_cause": "...",
    "contributing_factors": ["...", "..."],
    "confidence": "high" | "medium" | "low"
  },
  "resolution_steps": [
    {
      "step": 1,
      "action": "...",
      "responsible_party": "carrier" | "shipper" | "internal_team",
      "priority": "critical" | "high" | "medium",
      "estimated_resolution_time": "..."
    }
  ],
  "carrier_performance_context": {
    "carrier_on_time_pct_this_lane": 0.0-1.0,
    "carrier_invoice_accuracy_pct": 0.0-1.0,
    "is_recurring_issue": true/false
  },
  "financial_impact": {
    "potential_overcharge": 0.00,
    "potential_claim_value": 0.00
  }
}

Use Knowledge Base to query carrier performance history and contract terms.

Tools Enabled:

• Knowledge Base Tool (to retrieve carrier contracts, performance history, lane-specific data)
• Function Block (to calculate financial impact, compare rates)

Example Output:

{
  "root_cause_analysis": {
    "primary_cause": "Carrier consistently late on Chicago-to-Denver lane during January (winter weather delays)",
    "contributing_factors": [
      "I-80 corridor experiences heavy snow January-February",
      "Carrier does not adjust ETAs proactively for weather",
      "This is 4th delay on this lane in 30 days"
    ],
    "confidence": "high"
  },
  "resolution_steps": [
    {
      "step": 1,
      "action": "Contact carrier to understand if they need to adjust standard transit time for this lane during winter months",
      "responsible_party": "internal_team",
      "priority": "medium",
      "estimated_resolution_time": "24 hours"
    },
    {
      "step": 2,
      "action": "Consider adding 1-day buffer to expected delivery for Chicago-Denver shipments during Jan-Feb",
      "responsible_party": "internal_team",
      "priority": "medium",
      "estimated_resolution_time": "Immediate (update TMS settings)"
    },
    {
      "step": 3,
      "action": "Evaluate alternative carriers for this lane with better winter performance",
      "responsible_party": "internal_team",
      "priority": "low",
      "estimated_resolution_time": "1 week (requires RFP)"
    }
  ],
  "carrier_performance_context": {
    "carrier_on_time_pct_this_lane": 0.72,
    "carrier_invoice_accuracy_pct": 0.94,
    "is_recurring_issue": true
  },
  "financial_impact": {
    "potential_overcharge": 0.00,
    "potential_claim_value": 0.00
  }
}

Step 4: Automated Resolution & Communication

Use Agent Block + Mail Tool + Slack Tool to resolve exceptions automatically.

Resolution Workflow by Exception Type

For Delivery Delays:

Automated Actions (Function Block + Mail Tool):

def resolve_delivery_delay(shipment, exception, root_cause):
    """
    Automatically handle delivery delay resolution.
    """
    actions_taken = []

    # 1. Notify warehouse of updated ETA
    if exception['hours_late'] > 4 or exception['type'] == 'predicted_delay':
        send_email(
            to=shipment['destination']['warehouse_email'],
            subject=f"Updated Delivery ETA - Shipment {shipment['shipment_id']}",
            body=f"""
            The expected delivery for shipment {shipment['shipment_id']} has been updated.

            Original ETA: {shipment['expected_delivery_date']}
            New ETA: {shipment['carrier_estimated_delivery']}
            Delay: {exception.get('hours_late', exception.get('predicted_delay_hours'))} hours

            Carrier: {shipment['carrier']}
            Tracking: {shipment['tracking_number']}

            Root Cause: {root_cause['primary_cause']}

            Please adjust receiving schedule accordingly.
            """
        )
        actions_taken.append('Notified warehouse of updated ETA')

    # 2. Notify customer service if customer-facing shipment
    if shipment.get('customer_facing'):
        send_email(
            to='customer-service@company.com',
            subject=f"Customer Shipment Delayed - {shipment['shipment_id']}",
            body=f"""
            Customer shipment is delayed. Recommend proactive customer outreach.

            Order: {shipment.get('order_number')}
            Customer: {shipment.get('customer_name')}
            New Delivery Date: {shipment['carrier_estimated_delivery']}

            Suggested customer message template attached.
            """
        )
        actions_taken.append('Alerted customer service team')

    # 3. Escalate to logistics manager if delay exceeds threshold
    if exception['severity'] == 'critical':
        send_slack_message(
            channel='#logistics-exceptions',
            message=f"""
            🚨 Critical Delivery Delay

            Shipment: {shipment['shipment_id']}
            Carrier: {shipment['carrier']}
            Hours Late: {exception.get('hours_late', 'TBD')}

            Root Cause: {root_cause['primary_cause']}

            Recommended Actions:
            {chr(10).join(f"• {step['action']}" for step in root_cause['resolution_steps'])}

            <View Details>
            """
        )
        actions_taken.append('Escalated to logistics manager via Slack')

    return {
        'resolved': True if exception['severity'] != 'critical' else False,
        'requires_manual_review': exception['severity'] == 'critical',
        'actions_taken': actions_taken
    }

For Invoice Discrepancies:

Automated Validation & Dispute (Agent Block + Mail Tool):

def resolve_invoice_discrepancy(shipment, exception, carrier_contract):
    """
    Automatically validate invoice and create dispute if necessary.
    """
    # Calculate correct charge based on contract
    correct_weight = shipment['weight_lbs']  # Use TMS weight unless scale ticket available
    correct_rate = get_contracted_rate(carrier_contract, shipment['origin'], shipment['destination'], correct_weight)
    correct_total = correct_weight * correct_rate

    invoiced_total = shipment['invoice_amount']
    variance = invoiced_total - correct_total
    variance_pct = (variance / correct_total) * 100

    # If variance is <2%, auto-approve
    if abs(variance_pct) < 2:
        return {
            'action': 'auto_approved',
            'variance': round(variance, 2),
            'approved_amount': invoiced_total,
            'note': 'Variance within acceptable tolerance (<2%)'
        }

    # If variance >2%, flag for review and send dispute to carrier
    dispute_message = f"""
Dear {shipment['carrier']} Team,

We have identified a discrepancy on invoice #{shipment.get('invoice_number')} for shipment {shipment['shipment_id']}.

INVOICE DETAILS:
- Shipment ID: {shipment['shipment_id']}
- Tracking Number: {shipment['tracking_number']}
- Origin: {shipment['origin']['city']}, {shipment['origin']['state']}
- Destination: {shipment['destination']['city']}, {shipment['destination']['state']}

DISCREPANCY:
- Invoiced Weight: {shipment['invoice_weight']} lbs
- Our Records (TMS): {shipment['weight_lbs']} lbs
- Invoiced Rate: ${shipment['invoice_rate']}/lb
- Contracted Rate: ${correct_rate}/lb
- Invoiced Total: ${invoiced_total}
- Expected Total: ${correct_total}
- Variance: ${round(variance, 2)} ({round(variance_pct, 1)}%)

REQUESTED ACTION:
Please review and provide:
1. Documentation supporting the invoiced weight (scale ticket or carrier weight documentation)
2. Explanation of rate calculation (including any accessorials applied)
3. Corrected invoice if discrepancy is confirmed

Contract Reference: {carrier_contract['contract_id']}, Lane {shipment['origin']['zip'][:3]}-{shipment['destination']['zip'][:3]}

Best regards,
Automated Invoice Review System
    """

    # Send dispute email to carrier
    send_email(
        to=carrier_contract['billing_email'],
        cc='ap@company.com',
        subject=f"Invoice Dispute - {shipment['shipment_id']} - ${round(abs(variance), 2)} variance",
        body=dispute_message
    )

    # Create dispute record in TMS via API
    create_invoice_dispute(
        invoice_id=shipment['invoice_id'],
        variance_amount=variance,
        status='pending_carrier_response',
        dispute_reason=exception['description']
    )

    return {
        'action': 'disputed',
        'variance': round(variance, 2),
        'dispute_sent_to': carrier_contract['billing_email'],
        'status': 'pending_carrier_response'
    }

For Documentation Gaps:

Automated Requests (Mail Tool):

def resolve_documentation_gap(shipment, exception):
    """
    Automatically request missing documentation from responsible parties.
    """
    missing_docs = exception['missing_documents']
    requests_sent = []

    for doc_type in missing_docs:
        responsible_party = determine_responsible_party(doc_type, shipment)

        if responsible_party == 'carrier':
            send_email(
                to=shipment['carrier_contact_email'],
                subject=f"Missing {doc_type.upper()} - Shipment {shipment['shipment_id']}",
                body=f"""
Dear {shipment['carrier']} Team,

We are missing the {doc_type.replace('_', ' ').title()} for shipment {shipment['shipment_id']}.

Shipment Details:
- Tracking Number: {shipment['tracking_number']}
- Delivery Date: {shipment.get('actual_delivery_date', 'In transit')}

Please provide the {doc_type.replace('_', ' ')} within 24 hours.

Upload here: [Document Upload Portal]

Thank you,
Logistics Team
                """
            )
            requests_sent.append(f"{doc_type} requested from {shipment['carrier']}")

        elif responsible_party == 'shipper':
            # Notify internal shipper contact
            send_email(
                to=shipment['shipper_email'],
                subject=f"Action Required: Missing {doc_type.upper()} for Shipment {shipment['shipment_id']}",
                body=f"Please provide {doc_type.replace('_', ' ')} for shipment {shipment['shipment_id']}..."
            )
            requests_sent.append(f"{doc_type} requested from shipper")

    return {
        'action': 'documentation_requested',
        'requests_sent': requests_sent,
        'follow_up_date': (datetime.now() + timedelta(days=1)).isoformat()
    }

Step 5: Performance Insights & Trend Analysis

Use Function Block + Google Sheets Tool to track carrier performance and exception trends.

Carrier Performance Dashboard

def update_carrier_performance_dashboard(shipments_last_30_days, exceptions_last_30_days):
    """
    Calculate carrier performance metrics and update Google Sheets dashboard.
    """
    carrier_metrics = {}

    for shipment in shipments_last_30_days:
        carrier = shipment['carrier']

        if carrier not in carrier_metrics:
            carrier_metrics[carrier] = {
                'total_shipments': 0,
                'on_time': 0,
                'delayed': 0,
                'invoice_disputes': 0,
                'total_exceptions': 0,
                'avg_delay_hours': [],
                'total_spend': 0
            }

        carrier_metrics[carrier]['total_shipments'] += 1
        carrier_metrics[carrier]['total_spend'] += shipment.get('invoice_amount', 0)

        # Check if this shipment had a delay exception
        delay_exception = next((e for e in exceptions_last_30_days
                               if e['shipment_id'] == shipment['shipment_id'] and e['type'] == 'delivery_delay'), None)

        if delay_exception:
            carrier_metrics[carrier]['delayed'] += 1
            carrier_metrics[carrier]['avg_delay_hours'].append(delay_exception['hours_late'])
        else:
            carrier_metrics[carrier]['on_time'] += 1

        # Count invoice disputes
        invoice_exception = next((e for e in exceptions_last_30_days
                                 if e['shipment_id'] == shipment['shipment_id'] and 'invoice' in e['type']), None)

        if invoice_exception:
            carrier_metrics[carrier]['invoice_disputes'] += 1

        # Count total exceptions
        shipment_exceptions = [e for e in exceptions_last_30_days if e['shipment_id'] == shipment['shipment_id']]
        carrier_metrics[carrier]['total_exceptions'] += len(shipment_exceptions)

    # Calculate summary metrics
    dashboard_data = []

    for carrier, metrics in carrier_metrics.items():
        on_time_pct = (metrics['on_time'] / metrics['total_shipments']) * 100 if metrics['total_shipments'] > 0 else 0
        avg_delay = sum(metrics['avg_delay_hours']) / len(metrics['avg_delay_hours']) if metrics['avg_delay_hours'] else 0
        invoice_accuracy_pct = ((metrics['total_shipments'] - metrics['invoice_disputes']) / metrics['total_shipments']) * 100
        exception_rate = (metrics['total_exceptions'] / metrics['total_shipments']) * 100

        dashboard_data.append({
            'Carrier': carrier,
            'Total Shipments': metrics['total_shipments'],
            'On-Time %': round(on_time_pct, 1),
            'Avg Delay (hours)': round(avg_delay, 1),
            'Invoice Accuracy %': round(invoice_accuracy_pct, 1),
            'Exception Rate %': round(exception_rate, 1),
            'Total Spend': f"${metrics['total_spend']:,.2f}"
        })

    # Update Google Sheets
    update_google_sheet(
        spreadsheet_id='your-spreadsheet-id',
        range='Carrier Performance!A2:G100',
        values=[list(row.values()) for row in dashboard_data]
    )

    return dashboard_data

Exception Trend Analysis

def analyze_exception_trends(exceptions_last_90_days):
    """
    Identify patterns and trends in exceptions.
    """
    trends = {
        'by_type': {},
        'by_carrier': {},
        'by_lane': {},
        'by_day_of_week': {i: 0 for i in range(7)},
        'trending_up': [],
        'trending_down': []
    }

    for exception in exceptions_last_90_days:
        # Count by type
        exc_type = exception['type']
        if exc_type not in trends['by_type']:
            trends['by_type'][exc_type] = 0
        trends['by_type'][exc_type] += 1

        # Count by carrier
        carrier = exception['carrier']
        if carrier not in trends['by_carrier']:
            trends['by_carrier'][carrier] = 0
        trends['by_carrier'][carrier] += 1

        # Count by lane (origin-destination pair)
        lane = f"{exception['origin_city']}-{exception['dest_city']}"
        if lane not in trends['by_lane']:
            trends['by_lane'][lane] = 0
        trends['by_lane'][lane] += 1

        # Count by day of week
        exception_date = datetime.fromisoformat(exception['timestamp'])
        day_of_week = exception_date.weekday()
        trends['by_day_of_week'][day_of_week] += 1

    # Identify top problem areas
    top_problem_types = sorted(trends['by_type'].items(), key=lambda x: x[1], reverse=True)[:3]
    top_problem_carriers = sorted(trends['by_carrier'].items(), key=lambda x: x[1], reverse=True)[:3]
    top_problem_lanes = sorted(trends['by_lane'].items(), key=lambda x: x[1], reverse=True)[:5]

    return {
        'summary': {
            'total_exceptions': len(exceptions_last_90_days),
            'top_exception_types': top_problem_types,
            'top_problem_carriers': top_problem_carriers,
            'top_problem_lanes': top_problem_lanes
        },
        'recommendations': generate_recommendations(trends)
    }

def generate_recommendations(trends):
    """
    Generate actionable recommendations based on exception trends.
    """
    recommendations = []

    # Recommendation 1: Carrier performance
    worst_carrier = max(trends['by_carrier'].items(), key=lambda x: x[1])
    recommendations.append({
        'type': 'carrier_review',
        'priority': 'high',
        'recommendation': f"Review contract with {worst_carrier[0]} - {worst_carrier[1]} exceptions in 90 days",
        'action': f"Schedule quarterly business review with {worst_carrier[0]} to address performance issues"
    })

    # Recommendation 2: Lane-specific issues
    worst_lane = max(trends['by_lane'].items(), key=lambda x: x[1])
    recommendations.append({
        'type': 'lane_optimization',
        'priority': 'medium',
        'recommendation': f"Lane {worst_lane[0]} has {worst_lane[1]} exceptions - investigate root cause",
        'action': f"Evaluate alternative carriers or routing options for {worst_lane[0]} lane"
    })

    # Recommendation 3: Day-of-week patterns
    worst_day = max(trends['by_day_of_week'].items(), key=lambda x: x[1])
    day_names = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
    recommendations.append({
        'type': 'scheduling_optimization',
        'priority': 'low',
        'recommendation': f"{day_names[worst_day[0]]} has highest exception rate - consider schedule adjustments",
        'action': f"Avoid shipping on {day_names[worst_day[0]]} for time-sensitive shipments"
    })

    return recommendations

Step 6: Complete Klyntos Workflow Integration

Workflow Configuration

Trigger: Webhook receives shipment tracking update OR scheduled daily batch run

Workflow Steps:

1. WEBHOOK BLOCK: Receive carrier tracking update
   ↓
2. HTTP REQUEST TOOL: Fetch shipment details from TMS
   ↓
3. HTTP REQUEST TOOL: Fetch invoice data (if available)
   ↓
4. FUNCTION BLOCK: Create unified shipment view
   ↓
5. FUNCTION BLOCK: Detect exceptions
   ↓
6. CONDITION BLOCK: Route based on exception severity
   ↓
   ├─ CRITICAL →
   │   ├─ AGENT BLOCK: Analyze root cause
   │   ├─ FUNCTION BLOCK: Execute automated resolution
   │   ├─ SLACK TOOL: Send alert to logistics manager
   │   └─ MAIL TOOL: Notify affected parties
   │
   ├─ WARNING →
   │   ├─ AGENT BLOCK: Analyze root cause
   │   ├─ FUNCTION BLOCK: Execute automated resolution
   │   └─ MAIL TOOL: Notify coordinator
   │
   └─ INFO →
       └─ FUNCTION BLOCK: Log for trend analysis
   ↓
7. GOOGLE SHEETS TOOL: Update performance dashboard
   ↓
8. RESPONSE BLOCK: Return resolution summary

Example Klyntos Workflow (JSON Configuration)

{
  "workflow_name": "Supply Chain Exception Management",
  "trigger": {
    "type": "webhook",
    "endpoint": "/webhooks/carrier-tracking"
  },
  "blocks": [
    {
      "id": "fetch_tms_data",
      "type": "http_request",
      "config": {
        "method": "GET",
        "url": "https://tms.company.com/api/shipments/{{webhook.shipment_id}}",
        "headers": {
          "Authorization": "Bearer {{env.TMS_API_KEY}}"
        }
      }
    },
    {
      "id": "create_unified_view",
      "type": "function",
      "config": {
        "code": "create_unified_shipment_view(tms_data, carrier_tracking, invoice_data)"
      }
    },
    {
      "id": "detect_exceptions",
      "type": "function",
      "config": {
        "code": "detect_shipment_exceptions(shipment, carrier_contract)"
      }
    },
    {
      "id": "route_by_severity",
      "type": "condition",
      "config": {
        "conditions": [
          {
            "if": "{{exceptions.length > 0 && exceptions[0].severity == 'critical'}}",
            "then": "critical_path"
          },
          {
            "if": "{{exceptions.length > 0 && exceptions[0].severity == 'warning'}}",
            "then": "warning_path"
          }
        ],
        "else": "info_path"
      }
    },
    {
      "id": "analyze_root_cause",
      "type": "agent",
      "path": "critical_path",
      "config": {
        "model": "claude-3-5-sonnet-20241022",
        "system_prompt": "[Root cause analysis prompt from Step 3]",
        "tools": ["knowledge_base", "function"]
      }
    },
    {
      "id": "send_slack_alert",
      "type": "slack",
      "path": "critical_path",
      "config": {
        "channel": "#logistics-exceptions",
        "message": "🚨 Critical Exception: {{exception.description}}\nShipment: {{shipment.shipment_id}}\nRoot Cause: {{root_cause.primary_cause}}"
      }
    },
    {
      "id": "update_dashboard",
      "type": "google_sheets",
      "config": {
        "spreadsheet_id": "{{env.PERFORMANCE_DASHBOARD_ID}}",
        "range": "Exceptions!A:F",
        "append_row": true,
        "values": ["{{shipment.shipment_id}}", "{{exception.type}}", "{{exception.severity}}", "{{timestamp}}", "{{resolution.action}}", "{{root_cause.primary_cause}}"]
      }
    }
  ]
}

Testing Your Workflow

Test Case 1: Delivery Delay with Critical Severity

Input (Webhook Payload):

{
  "shipment_id": "SHP-12345",
  "status": "delivered",
  "actual_delivery": "2025-01-17T22:00:00Z",
  "expected_delivery": "2025-01-16T16:00:00Z"
}

Expected Output:

• Exception detected: delivery_delay, 30 hours late, severity: critical
• Root cause: Carrier routing error + winter weather delay on I-80
• Actions taken:
  • Warehouse notified of delay
  • Customer service alerted
  • Slack alert sent to logistics manager
  • Carrier contacted for explanation
• Dashboard updated with exception record

Test Case 2: Invoice Discrepancy

Input:

{
  "invoice_id": "INV-98765",
  "shipment_id": "SHP-23456",
  "invoice_weight": 1350,
  "tms_weight": 1200,
  "invoice_rate": 4.25,
  "contracted_rate": 3.75
}

Expected Output:

• Exception detected: weight_discrepancy (12.5% variance) + invoice_rate_anomaly (13.3% variance)
• Calculated overcharge: $675 ($1.50 overcharge on 450 lb difference)
• Actions taken:
  • Dispute email sent to carrier billing team
  • Invoice flagged for AP review
  • Dispute record created in TMS
• Status: pending_carrier_response

Test Case 3: Temperature Violation (Food Distributor)

Input:

{
  "shipment_id": "SHP-34567",
  "special_handling": ["temperature_controlled"],
  "temperature_events": [
    {
      "timestamp": "2025-01-16T08:00:00Z",
      "temp_f": 45,
      "min_temp": 32,
      "max_temp": 38
    }
  ]
}

Expected Output:

• Exception detected: temperature_violation, severity: critical
• Root cause: Refrigeration unit failure or door left open
• Actions taken:
  • Immediate Slack alert to quality team
  • Carrier contacted to investigate
  • Shipment flagged for inspection upon receipt
  • Potential spoilage claim initiated
• Status: requires_manual_inspection

Test Case 4: Batch Exception Analysis (Daily Report)

Input:

• 500 shipments processed in last 24 hours
• 47 exceptions detected (9.4% exception rate)

Expected Output:

• Carrier performance updated in Google Sheets
• Top 3 exception types: delivery_delay (28), invoice_rate_anomaly (12), documentation_gap (7)
• Worst performing carrier: Carrier XYZ (18 exceptions out of 120 shipments)
• Worst performing lane: Chicago-Denver (9 exceptions)
• Recommendations generated:
  1. Review Carrier XYZ performance and consider contract renegotiation
  2. Investigate Chicago-Denver lane for systematic issues
  3. Implement additional documentation checks for international shipments

Advanced Enhancements

1. Predictive Delay Modeling

Use historical data to predict delays before they happen:

def predict_delivery_delay(shipment, historical_data):
    """
    Predict likelihood of delay based on historical patterns.

    Factors:
    - Carrier on-time performance for this lane
    - Weather forecast along route
    - Day of week and seasonality
    - Current carrier load
    """
    pass

2. Dynamic Carrier Routing

Automatically select best carrier for each shipment based on performance:

def recommend_carrier(shipment_requirements, carrier_performance):
    """
    Recommend optimal carrier based on:
    - On-time performance for this lane
    - Invoice accuracy
    - Cost
    - Capacity availability
    """
    pass

3. Automated Claims Filing

For temperature violations or damaged shipments, automatically initiate claims:

def file_carrier_claim(shipment, exception, damage_amount):
    """
    Automatically file claim with carrier for:
    - Temperature violations causing spoilage
    - Lost or damaged shipments
    - Service failures (late delivery fees)
    """
    pass

4. Real-Time Customer Notifications

Integrate with customer notification system to proactively communicate delays.

ROI Metrics

Metric	Before Klyntos	After Klyntos	Improvement
Exception management time per week	40 hours	7 hours	82% reduction
On-time delivery rate	87%	98%	13% improvement
Invoice dispute rate	15%	3%	80% reduction
Temperature compliance (food)	89%	99%	10% improvement
Documentation completeness	76%	98%	22% improvement
Carrier cost savings	Baseline	-3%	Cost reduction
Volume handled (same team)	1x	2.5x	150% increase

Troubleshooting

Issue: Webhook not receiving carrier tracking updates Solution: Verify webhook URL is registered with carrier API. Check authentication credentials. Test with carrier's webhook testing tool.

Issue: False positive on delivery delays (shipment marked late but actually on time) Solution: Ensure TMS expected delivery date matches carrier's committed delivery date. Account for weekend/holiday adjustments in delay calculations.

Issue: Invoice rate lookups failing (can't find contracted rate) Solution: Verify carrier contract is properly loaded into Knowledge Base. Ensure lane matching logic handles zip code prefixes correctly. Add logging to see which lanes are not matching.

Issue: Too many false alerts on invoice discrepancies Solution: Adjust variance thresholds (currently 5%) based on your tolerance. Consider excluding certain accessorials (fuel surcharges) from variance calculations if they're expected to fluctuate.

---

Next Steps

1. Connect your systems - Integrate TMS, carrier APIs, and invoice platforms
2. Define exception thresholds - Determine what constitutes critical vs warning exceptions for your operations
3. Configure notification workflows - Set up Slack channels and email distribution lists
4. Test with historical data - Run past shipments through the system to validate detection accuracy
5. Train your team - Show coordinators how to use the exception dashboard and respond to automated alerts

For implementation support, visit the Klyntos documentation or contact our team.