Building a Supply Chain Visibility Agent: End-to-End Shipment Tracking and Alerts

The Supply Chain Visibility Problem

A single product might travel by truck from a factory to a port, by ocean vessel across the Pacific, by rail from the port to a distribution center, and by truck again for final delivery. Each leg involves a different carrier, a different tracking system, and different milestone events. Supply chain managers today toggle between five or more carrier portals, spreadsheets, and email threads to piece together where their goods are.

An AI visibility agent aggregates tracking data across all transport modes into a single timeline, predicts delays before they happen, and proactively notifies stakeholders when milestones are met or disruptions occur.

Multi-Modal Shipment Data Model

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional

class TransportMode(str, Enum):
    OCEAN = "ocean"
    AIR = "air"
    RAIL = "rail"
    TRUCK = "truck"

class MilestoneStatus(str, Enum):
    COMPLETED = "completed"
    IN_PROGRESS = "in_progress"
    PENDING = "pending"
    DELAYED = "delayed"
    EXCEPTION = "exception"

@dataclass
class Milestone:
    name: str
    mode: TransportMode
    location: str
    planned_date: datetime
    actual_date: Optional[datetime] = None
    status: MilestoneStatus = MilestoneStatus.PENDING
    carrier: str = ""
    reference: str = ""

@dataclass
class SupplyChainShipment:
    shipment_id: str
    po_number: str
    origin_country: str
    destination: str
    product: str
    quantity: int
    milestones: list[Milestone] = field(default_factory=list)
    stakeholders: list[dict] = field(default_factory=list)

Sample Shipment Data

SHIPMENTS = {
    "SC-70001": SupplyChainShipment(
        shipment_id="SC-70001",
        po_number="PO-2026-1234",
        origin_country="China",
        destination="Chicago, IL",
        product="Electronic Components",
        quantity=5000,
        milestones=[
            Milestone("Factory Pickup", TransportMode.TRUCK, "Shenzhen",
                      datetime(2026, 3, 1, 8, 0), datetime(2026, 3, 1, 9, 30),
                      MilestoneStatus.COMPLETED, "Local Trucking Co", "TRK-001"),
            Milestone("Port Departure", TransportMode.OCEAN, "Yantian Port",
                      datetime(2026, 3, 3, 6, 0), datetime(2026, 3, 3, 14, 0),
                      MilestoneStatus.COMPLETED, "COSCO", "COSU-1234567"),
            Milestone("Port Arrival", TransportMode.OCEAN, "Long Beach, CA",
                      datetime(2026, 3, 17, 8, 0), None,
                      MilestoneStatus.IN_PROGRESS, "COSCO", "COSU-1234567"),
            Milestone("Customs Clearance", TransportMode.TRUCK, "Long Beach, CA",
                      datetime(2026, 3, 18, 12, 0), None,
                      MilestoneStatus.PENDING, "Customs Broker LLC", "CB-5678"),
            Milestone("Rail Departure", TransportMode.RAIL, "Long Beach, CA",
                      datetime(2026, 3, 19, 6, 0), None,
                      MilestoneStatus.PENDING, "BNSF", "BNSF-9876"),
            Milestone("Rail Arrival", TransportMode.RAIL, "Chicago, IL",
                      datetime(2026, 3, 22, 10, 0), None,
                      MilestoneStatus.PENDING, "BNSF", "BNSF-9876"),
            Milestone("Final Delivery", TransportMode.TRUCK, "Chicago, IL",
                      datetime(2026, 3, 23, 14, 0), None,
                      MilestoneStatus.PENDING, "XPO Logistics", "XPO-4321"),
        ],
        stakeholders=[
            {"name": "Procurement Team", "email": "procurement@example.com", "role": "buyer"},
            {"name": "Warehouse Ops", "email": "warehouse@example.com", "role": "receiver"},
            {"name": "Sales Team", "email": "sales@example.com", "role": "downstream"},
        ],
    ),
}

Shipment Tracking Tool

from agents import function_tool

@function_tool
def track_shipment(
    shipment_id: Optional[str] = None,
    po_number: Optional[str] = None,
) -> str:
    """Track a supply chain shipment by ID or PO number with full milestone timeline."""
    shipment = None
    if shipment_id:
        shipment = SHIPMENTS.get(shipment_id)
    elif po_number:
        shipment = next(
            (s for s in SHIPMENTS.values() if s.po_number == po_number), None
        )

    if not shipment:
        return "Shipment not found. Please check the ID or PO number."

    lines = [
        f"=== Shipment {shipment.shipment_id} ===",
        f"PO: {shipment.po_number}",
        f"Product: {shipment.product} (qty: {shipment.quantity})",
        f"Route: {shipment.origin_country} -> {shipment.destination}\n",
        "Milestone Timeline:",
    ]

    for m in shipment.milestones:
        status_icon = {
            MilestoneStatus.COMPLETED: "DONE",
            MilestoneStatus.IN_PROGRESS: "ACTIVE",
            MilestoneStatus.PENDING: "PENDING",
            MilestoneStatus.DELAYED: "DELAYED",
            MilestoneStatus.EXCEPTION: "EXCEPTION",
        }[m.status]

        planned = m.planned_date.strftime("%m/%d %H:%M")
        actual = m.actual_date.strftime("%m/%d %H:%M") if m.actual_date else "---"

        delay_note = ""
        if m.actual_date and m.actual_date > m.planned_date:
            hours_late = (m.actual_date - m.planned_date).total_seconds() / 3600
            delay_note = f" (+{hours_late:.0f}h late)"

        lines.append(
            f"  [{status_icon}] {m.name} ({m.mode.value}) @ {m.location}\n"
            f"         Planned: {planned} | Actual: {actual}{delay_note}\n"
            f"         Carrier: {m.carrier} | Ref: {m.reference}"
        )

    return "\n".join(lines)

Delay Prediction Tool

The delay predictor analyzes current milestone performance to estimate downstream impact:

@function_tool
def predict_delays(shipment_id: str) -> str:
    """Predict potential delays for a shipment based on current milestone performance."""
    shipment = SHIPMENTS.get(shipment_id)
    if not shipment:
        return "Shipment not found."

    # Calculate cumulative delay from completed milestones
    total_delay_hours = 0.0
    for m in shipment.milestones:
        if m.actual_date and m.actual_date > m.planned_date:
            total_delay_hours += (m.actual_date - m.planned_date).total_seconds() / 3600

    # Find current active milestone
    active = next(
        (m for m in shipment.milestones
         if m.status == MilestoneStatus.IN_PROGRESS), None
    )

    predictions = []

    if total_delay_hours > 0:
        predictions.append(
            f"Cumulative delay so far: {total_delay_hours:.0f} hours"
        )

        # Predict impact on pending milestones
        for m in shipment.milestones:
            if m.status == MilestoneStatus.PENDING:
                # Simple propagation: delay carries forward minus buffer
                buffer_hours = 4.0 if m.mode == TransportMode.RAIL else 2.0
                predicted_delay = max(0, total_delay_hours - buffer_hours)
                if predicted_delay > 0:
                    predictions.append(
                        f"  {m.name}: likely {predicted_delay:.0f}h late "
                        f"(original: {m.planned_date.strftime('%m/%d %H:%M')})"
                    )

        # Check if final delivery is at risk
        final = shipment.milestones[-1]
        if total_delay_hours > 8:
            predictions.append(
                f"\nWARNING: Final delivery to {shipment.destination} "
                f"is at risk of missing the planned window."
            )
    else:
        predictions.append("No delays detected. Shipment is on schedule.")

    return "\n".join(predictions)

Stakeholder Notification Tool

@function_tool
def notify_stakeholders(
    shipment_id: str,
    message: str,
    roles: Optional[list[str]] = None,
    priority: str = "normal",
) -> str:
    """Send notifications to shipment stakeholders by role."""
    shipment = SHIPMENTS.get(shipment_id)
    if not shipment:
        return "Shipment not found."

    recipients = shipment.stakeholders
    if roles:
        recipients = [s for s in recipients if s["role"] in roles]

    if not recipients:
        return "No matching stakeholders found."

    notifications = [f"Notifications sent for {shipment_id} [{priority.upper()}]:"]
    for r in recipients:
        notifications.append(
            f"  -> {r['name']} ({r['role']}): {r['email']}"
        )
    notifications.append(f"\nMessage: {message}")

    return "\n".join(notifications)

Assembling the Visibility Agent

from agents import Agent, Runner

visibility_agent = Agent(
    name="Supply Chain Visibility",
    instructions="""You are a supply chain visibility assistant. Help logistics teams:
    1. Track shipments end-to-end across ocean, air, rail, and truck
    2. Predict delays based on current milestone performance
    3. Notify stakeholders proactively about status changes and delays
    Always explain delays in business impact terms (e.g., warehouse receiving impact).""",
    tools=[track_shipment, predict_delays, notify_stakeholders],
)

result = Runner.run_sync(
    visibility_agent,
    "What's the status of PO-2026-1234? Are there any predicted delays? "
    "If so, notify the warehouse team."
)
print(result.final_output)

FAQ

How do I aggregate data from real carriers across different transport modes?

Use supply chain visibility platforms like project44, FourKites, or Chain.io which aggregate tracking data across ocean (via AIS and carrier EDI), rail (Class I railroad APIs), and truck (ELD/GPS). These platforms normalize events into standard milestone formats. Subscribe to webhook events for real-time updates rather than polling.

How accurate can delay predictions be?

Simple delay propagation like shown here works for basic cascading delays. For higher accuracy, build a machine learning model trained on historical shipment data for your specific lanes. Features include origin port congestion, vessel schedule reliability, customs clearance times by commodity code, and seasonal patterns. Even a gradient-boosted model on 12 months of data can significantly outperform carrier ETAs.

How should the agent handle force majeure events like port strikes or natural disasters?

Build a disruption monitoring tool that checks news feeds, port status APIs, and weather services. When a disruption is detected in a region that affects active shipments, the agent should proactively identify all impacted shipments, estimate the delay, and notify stakeholders with recommended actions like rerouting or expediting alternative transport modes.

---

#SupplyChain #ShipmentVisibility #MultiModalTracking #DelayPrediction #Python #AgenticAI #LearnAI #AIEngineering