--- title: "Building a Supply Chain Visibility Agent: End-to-End Shipment Tracking and Alerts" description: "Build an AI agent that provides end-to-end supply chain visibility across ocean, air, rail, and truck shipments with milestone tracking, delay prediction, and automated stakeholder notifications." canonical: https://callsphere.ai/blog/building-supply-chain-visibility-agent-end-to-end-shipment-tracking-alerts category: "Learn Agentic AI" tags: ["Supply Chain", "Shipment Visibility", "Multi-Modal Tracking", "Delay Prediction", "Python"] author: "CallSphere Team" published: 2026-03-17T00:00:00.000Z updated: 2026-05-06T01:02:44.312Z --- # Building a Supply Chain Visibility Agent: End-to-End Shipment Tracking and Alerts > Build an AI agent that provides end-to-end supply chain visibility across ocean, air, rail, and truck shipments with milestone tracking, delay prediction, and automated stakeholder notifications. ## 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 ```python 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 ```python 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 ```python 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: ```mermaid flowchart LR REL(["Release of Building a Supply Chain Visibility A"]) NEW1["What's new flagship feature 1"] NEW2["What's new flagship feature 2"] NEW3["What's new flagship feature 3"] BREAK{"Breaking changes?"} MIG["Migration steps"] UPG(["Upgrade now"]) WAIT(["Pin current, upgrade later"]) REL --> NEW1 REL --> NEW2 REL --> NEW3 NEW1 --> BREAK NEW2 --> BREAK NEW3 --> BREAK BREAK -->|Yes| MIG --> UPG BREAK -->|No| UPG BREAK -->|Risk averse| WAIT style REL fill:#4f46e5,stroke:#4338ca,color:#fff style BREAK fill:#f59e0b,stroke:#d97706,color:#1f2937 style UPG fill:#059669,stroke:#047857,color:#fff style WAIT fill:#0ea5e9,stroke:#0369a1,color:#fff ``` ```python @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 ```python @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 ```python 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 --- Source: https://callsphere.ai/blog/building-supply-chain-visibility-agent-end-to-end-shipment-tracking-alerts