feat: Implement logistics agents and adapters for carrier selection, ETA computation, returns support, and route issue detection

Author: Cataldir

apps/logistics-carrier-selection/src/logistics_carrier_selection/adapters.py

@@ -0,0 +1,44 @@
+"""Adapters for the logistics carrier selection service."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Optional
+
+from holiday_peak_lib.adapters.logistics_adapter import LogisticsConnector
+from holiday_peak_lib.adapters.mock_adapters import MockLogisticsAdapter
+from holiday_peak_lib.schemas.logistics import LogisticsContext
+
+
+@dataclass
+class CarrierSelectionAdapters:
+    """Container for carrier selection adapters."""
+
+    logistics: LogisticsConnector
+    selector: "CarrierSelector"
+
+
+class CarrierSelector:
+    """Choose a carrier based on shipment attributes."""
+
+    async def select(self, context: LogisticsContext) -> dict[str, Any]:
+        shipment = context.shipment
+        service = shipment.service_level or "standard"
+        carrier = shipment.carrier or ("priority-carrier" if service == "express" else "economy-carrier")
+        return {
+            "tracking_id": shipment.tracking_id,
+            "service_level": service,
+            "recommended_carrier": carrier,
+            "reason": "matched service level",
+        }
+
+
+def build_carrier_selection_adapters(
+    *, logistics_connector: Optional[LogisticsConnector] = None
+) -> CarrierSelectionAdapters:
+    """Create adapters for carrier selection workflows.
+
+    Uses mock adapters by default to keep local development lightweight.
+    """
+    logistics = logistics_connector or LogisticsConnector(adapter=MockLogisticsAdapter())
+    selector = CarrierSelector()
+    return CarrierSelectionAdapters(logistics=logistics, selector=selector)

apps/logistics-carrier-selection/src/logistics_carrier_selection/agents.py

@@ -0,0 +1,86 @@
+"""Logistics carrier selection agent implementation and MCP tool registration."""
+from __future__ import annotations
+
+from typing import Any
+
+from holiday_peak_lib.agents import BaseRetailAgent
+from holiday_peak_lib.agents.fastapi_mcp import FastAPIMCPServer
+
+from .adapters import CarrierSelectionAdapters, build_carrier_selection_adapters
+
+
+class CarrierSelectionAgent(BaseRetailAgent):
+    """Agent that recommends a carrier for a shipment."""
+
+    def __init__(self, config, *args: Any, **kwargs: Any) -> None:
+        super().__init__(config, *args, **kwargs)
+        self._adapters = build_carrier_selection_adapters()
+
+    @property
+    def adapters(self) -> CarrierSelectionAdapters:
+        return self._adapters
+
+    async def handle(self, request: dict[str, Any]) -> dict[str, Any]:
+        tracking_id = request.get("tracking_id")
+        if not tracking_id:
+            return {"error": "tracking_id is required"}
+
+        context = await self.adapters.logistics.build_logistics_context(str(tracking_id))
+        if not context:
+            return {"error": "shipment not found", "tracking_id": tracking_id}
+
+        recommendation = await self.adapters.selector.select(context)
+
+        if self.slm or self.llm:
+            messages = [
+                {"role": "system", "content": _carrier_instructions()},
+                {
+                    "role": "user",
+                    "content": {
+                        "tracking_id": tracking_id,
+                        "shipment": context.model_dump(),
+                        "recommendation": recommendation,
+                    },
+                },
+            ]
+            return await self.invoke_model(request=request, messages=messages)
+
+        return {
+            "service": self.service_name,
+            "tracking_id": tracking_id,
+            "shipment": context.model_dump(),
+            "recommendation": recommendation,
+        }
+
+
+def register_mcp_tools(mcp: FastAPIMCPServer, agent: BaseRetailAgent) -> None:
+    """Expose MCP tools for carrier selection workflows."""
+    adapters = getattr(agent, "adapters", build_carrier_selection_adapters())
+
+    async def get_logistics_context(payload: dict[str, Any]) -> dict[str, Any]:
+        tracking_id = payload.get("tracking_id")
+        if not tracking_id:
+            return {"error": "tracking_id is required"}
+        context = await adapters.logistics.build_logistics_context(str(tracking_id))
+        return {"logistics_context": context.model_dump() if context else None}
+
+    async def get_carrier_recommendation(payload: dict[str, Any]) -> dict[str, Any]:
+        tracking_id = payload.get("tracking_id")
+        if not tracking_id:
+            return {"error": "tracking_id is required"}
+        context = await adapters.logistics.build_logistics_context(str(tracking_id))
+        if not context:
+            return {"error": "shipment not found", "tracking_id": tracking_id}
+        recommendation = await adapters.selector.select(context)
+        return {"recommendation": recommendation}
+
+    mcp.add_tool("/logistics/carrier/context", get_logistics_context)
+    mcp.add_tool("/logistics/carrier/recommendation", get_carrier_recommendation)
+
+
+def _carrier_instructions() -> str:
+    return (
+        "You are a logistics carrier selection agent. "
+        "Recommend the best carrier based on service level and constraints. "
+        "Explain trade-offs and risks."
+    )

apps/logistics-carrier-selection/src/logistics_carrier_selection/main.py

@@ -1,14 +1,15 @@
 """Logistics carrier selection service."""
 from holiday_peak_lib.agents.memory import ColdMemory, HotMemory, WarmMemory
-from holiday_peak_lib.agents.service_agent import ServiceAgent
 from holiday_peak_lib.app_factory import build_service_app
 from holiday_peak_lib.config import MemorySettings
 
+from logistics_carrier_selection.agents import CarrierSelectionAgent, register_mcp_tools
+
 SERVICE_NAME = "logistics-carrier-selection"
 memory_settings = MemorySettings()
 app = build_service_app(
 	SERVICE_NAME,
-	agent_class=ServiceAgent,
+	agent_class=CarrierSelectionAgent,
 	hot_memory=HotMemory(memory_settings.redis_url),
 	warm_memory=WarmMemory(
 		memory_settings.cosmos_account_uri,
@@ -19,4 +20,5 @@
 		memory_settings.blob_account_url,
 		memory_settings.blob_container,
 	),
+	mcp_setup=register_mcp_tools,
 )

apps/logistics-eta-computation/src/logistics_eta_computation/adapters.py

@@ -0,0 +1,51 @@
+"""Adapters for the logistics ETA computation service."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timedelta
+from typing import Any, Optional
+
+from holiday_peak_lib.adapters.logistics_adapter import LogisticsConnector
+from holiday_peak_lib.adapters.mock_adapters import MockLogisticsAdapter
+from holiday_peak_lib.schemas.logistics import LogisticsContext
+
+
+@dataclass
+class EtaComputationAdapters:
+    """Container for ETA computation adapters."""
+
+    logistics: LogisticsConnector
+    estimator: "EtaEstimator"
+
+
+class EtaEstimator:
+    """Compute ETA based on event timeline."""
+
+    async def compute_eta(self, context: LogisticsContext) -> dict[str, Any]:
+        shipment = context.shipment
+        base_eta = shipment.eta
+        if base_eta:
+            return {
+                "tracking_id": shipment.tracking_id,
+                "eta": base_eta.isoformat(),
+                "source": "carrier",
+            }
+        now = datetime.utcnow()
+        inferred_eta = now + timedelta(days=2)
+        return {
+            "tracking_id": shipment.tracking_id,
+            "eta": inferred_eta.isoformat(),
+            "source": "estimated",
+        }
+
+
+def build_eta_adapters(
+    *, logistics_connector: Optional[LogisticsConnector] = None
+) -> EtaComputationAdapters:
+    """Create adapters for ETA computation workflows.
+
+    Uses mock adapters by default to keep local development lightweight.
+    """
+    logistics = logistics_connector or LogisticsConnector(adapter=MockLogisticsAdapter())
+    estimator = EtaEstimator()
+    return EtaComputationAdapters(logistics=logistics, estimator=estimator)

apps/logistics-eta-computation/src/logistics_eta_computation/agents.py

@@ -0,0 +1,86 @@
+"""Logistics ETA computation agent implementation and MCP tool registration."""
+from __future__ import annotations
+
+from typing import Any
+
+from holiday_peak_lib.agents import BaseRetailAgent
+from holiday_peak_lib.agents.fastapi_mcp import FastAPIMCPServer
+
+from .adapters import EtaComputationAdapters, build_eta_adapters
+
+
+class EtaComputationAgent(BaseRetailAgent):
+    """Agent that computes updated ETA for shipments."""
+
+    def __init__(self, config, *args: Any, **kwargs: Any) -> None:
+        super().__init__(config, *args, **kwargs)
+        self._adapters = build_eta_adapters()
+
+    @property
+    def adapters(self) -> EtaComputationAdapters:
+        return self._adapters
+
+    async def handle(self, request: dict[str, Any]) -> dict[str, Any]:
+        tracking_id = request.get("tracking_id")
+        if not tracking_id:
+            return {"error": "tracking_id is required"}
+
+        context = await self.adapters.logistics.build_logistics_context(str(tracking_id))
+        if not context:
+            return {"error": "shipment not found", "tracking_id": tracking_id}
+
+        eta = await self.adapters.estimator.compute_eta(context)
+
+        if self.slm or self.llm:
+            messages = [
+                {"role": "system", "content": _eta_instructions()},
+                {
+                    "role": "user",
+                    "content": {
+                        "tracking_id": tracking_id,
+                        "logistics_context": context.model_dump(),
+                        "eta": eta,
+                    },
+                },
+            ]
+            return await self.invoke_model(request=request, messages=messages)
+
+        return {
+            "service": self.service_name,
+            "tracking_id": tracking_id,
+            "logistics_context": context.model_dump(),
+            "eta": eta,
+        }
+
+
+def register_mcp_tools(mcp: FastAPIMCPServer, agent: BaseRetailAgent) -> None:
+    """Expose MCP tools for ETA computation workflows."""
+    adapters = getattr(agent, "adapters", build_eta_adapters())
+
+    async def get_logistics_context(payload: dict[str, Any]) -> dict[str, Any]:
+        tracking_id = payload.get("tracking_id")
+        if not tracking_id:
+            return {"error": "tracking_id is required"}
+        context = await adapters.logistics.build_logistics_context(str(tracking_id))
+        return {"logistics_context": context.model_dump() if context else None}
+
+    async def get_eta(payload: dict[str, Any]) -> dict[str, Any]:
+        tracking_id = payload.get("tracking_id")
+        if not tracking_id:
+            return {"error": "tracking_id is required"}
+        context = await adapters.logistics.build_logistics_context(str(tracking_id))
+        if not context:
+            return {"error": "shipment not found", "tracking_id": tracking_id}
+        eta = await adapters.estimator.compute_eta(context)
+        return {"eta": eta}
+
+    mcp.add_tool("/logistics/eta/context", get_logistics_context)
+    mcp.add_tool("/logistics/eta", get_eta)
+
+
+def _eta_instructions() -> str:
+    return (
+        "You are a logistics ETA computation agent. "
+        "Provide the most accurate ETA and explain confidence level. "
+        "Highlight delays and risk factors."
+    )

apps/logistics-eta-computation/src/logistics_eta_computation/main.py

@@ -1,14 +1,15 @@
 """Logistics ETA computation service."""
 from holiday_peak_lib.agents.memory import ColdMemory, HotMemory, WarmMemory
-from holiday_peak_lib.agents.service_agent import ServiceAgent
 from holiday_peak_lib.app_factory import build_service_app
 from holiday_peak_lib.config import MemorySettings
 
+from logistics_eta_computation.agents import EtaComputationAgent, register_mcp_tools
+
 SERVICE_NAME = "logistics-eta-computation"
 memory_settings = MemorySettings()
 app = build_service_app(
 	SERVICE_NAME,
-	agent_class=ServiceAgent,
+	agent_class=EtaComputationAgent,
 	hot_memory=HotMemory(memory_settings.redis_url),
 	warm_memory=WarmMemory(
 		memory_settings.cosmos_account_uri,
@@ -19,4 +20,5 @@
 		memory_settings.blob_account_url,
 		memory_settings.blob_container,
 	),
+	mcp_setup=register_mcp_tools,
 )

apps/logistics-returns-support/src/logistics_returns_support/adapters.py

@@ -0,0 +1,49 @@
+"""Adapters for the logistics returns support service."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Optional
+
+from holiday_peak_lib.adapters.logistics_adapter import LogisticsConnector
+from holiday_peak_lib.adapters.mock_adapters import MockLogisticsAdapter
+from holiday_peak_lib.schemas.logistics import LogisticsContext
+
+
+@dataclass
+class ReturnsSupportAdapters:
+    """Container for returns support adapters."""
+
+    logistics: LogisticsConnector
+    assistant: "ReturnsSupportAssistant"
+
+
+class ReturnsSupportAssistant:
+    """Generate returns support guidance from logistics context."""
+
+    async def build_returns_plan(self, context: LogisticsContext) -> dict[str, Any]:
+        shipment = context.shipment
+        status = shipment.status
+        return {
+            "tracking_id": shipment.tracking_id,
+            "status": status,
+            "eligible_for_return": status in {"delivered", "in_transit"},
+            "next_steps": _return_next_steps(status),
+        }
+
+
+def build_returns_support_adapters(
+    *, logistics_connector: Optional[LogisticsConnector] = None
+) -> ReturnsSupportAdapters:
+    """Create adapters for returns support workflows.
+
+    Uses mock adapters by default to keep local development lightweight.
+    """
+    logistics = logistics_connector or LogisticsConnector(adapter=MockLogisticsAdapter())
+    assistant = ReturnsSupportAssistant()
+    return ReturnsSupportAdapters(logistics=logistics, assistant=assistant)
+
+
+def _return_next_steps(status: str) -> list[str]:
+    if status == "delivered":
+        return ["Confirm return reason", "Issue return label", "Schedule pickup"]
+    return ["Confirm shipment status", "Offer return window", "Notify support"]