feat: Implement product agents and adapters for ACP transformation, assortment optimization, consistency validation, and normalization/classification

Author: Cataldir

apps/product-management-acp-transformation/src/product_management_acp_transformation/adapters.py

@@ -0,0 +1,62 @@
+"""Adapters for the product ACP transformation service."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Optional
+
+from holiday_peak_lib.adapters.mock_adapters import MockProductAdapter
+from holiday_peak_lib.adapters.product_adapter import ProductConnector
+from holiday_peak_lib.schemas.product import CatalogProduct
+
+
+@dataclass
+class AcpTransformationAdapters:
+    """Container for ACP transformation adapters."""
+
+    products: ProductConnector
+    mapper: "AcpCatalogMapper"
+
+
+class AcpCatalogMapper:
+    """Map catalog products to ACP Product Feed-like fields."""
+
+    def to_acp_product(
+        self,
+        product: CatalogProduct,
+        *,
+        availability: str,
+        currency: str = "usd",
+    ) -> dict[str, object]:
+        sku = product.sku
+        price = product.price if product.price is not None else 0.0
+        image_url = product.image_url or "https://example.com/images/placeholder.png"
+        product_url = f"https://example.com/products/{sku}"
+        return {
+            "item_id": sku,
+            "title": product.name,
+            "description": product.description or "",
+            "url": product_url,
+            "image_url": image_url,
+            "brand": product.brand or "",
+            "price": f"{price:.2f} {currency}",
+            "availability": availability,
+            "is_eligible_search": True,
+            "is_eligible_checkout": True,
+            "store_name": "Example Store",
+            "seller_url": "https://example.com/store",
+            "seller_privacy_policy": "https://example.com/privacy",
+            "seller_tos": "https://example.com/terms",
+            "return_policy": "https://example.com/returns",
+            "return_window": 30,
+            "target_countries": ["US"],
+            "store_country": "US",
+        }
+
+
+def build_acp_transformation_adapters(
+    *, product_connector: Optional[ProductConnector] = None
+) -> AcpTransformationAdapters:
+    """Create adapters for ACP transformation workflows."""
+    products = product_connector or ProductConnector(adapter=MockProductAdapter())
+    mapper = AcpCatalogMapper()
+    return AcpTransformationAdapters(products=products, mapper=mapper)

apps/product-management-acp-transformation/src/product_management_acp_transformation/agents.py

@@ -0,0 +1,94 @@
+"""Product ACP transformation 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 AcpTransformationAdapters, build_acp_transformation_adapters
+
+
+class ProductAcpTransformationAgent(BaseRetailAgent):
+    """Agent that transforms catalog products into ACP payloads."""
+
+    def __init__(self, config, *args: Any, **kwargs: Any) -> None:
+        super().__init__(config, *args, **kwargs)
+        self._adapters = build_acp_transformation_adapters()
+
+    @property
+    def adapters(self) -> AcpTransformationAdapters:
+        return self._adapters
+
+    async def handle(self, request: dict[str, Any]) -> dict[str, Any]:
+        sku = request.get("sku")
+        if not sku:
+            return {"error": "sku is required"}
+        availability = request.get("availability", "in_stock")
+        currency = request.get("currency", "usd")
+
+        product = await self.adapters.products.get_product(str(sku))
+        if not product:
+            return {"error": "sku not found", "sku": sku}
+
+        acp = self.adapters.mapper.to_acp_product(
+            product, availability=str(availability), currency=str(currency)
+        )
+
+        if self.slm or self.llm:
+            messages = [
+                {"role": "system", "content": _acp_instructions()},
+                {
+                    "role": "user",
+                    "content": {
+                        "sku": sku,
+                        "product": product.model_dump(),
+                        "acp_product": acp,
+                    },
+                },
+            ]
+            return await self.invoke_model(request=request, messages=messages)
+
+        return {
+            "service": self.service_name,
+            "sku": sku,
+            "product": product.model_dump(),
+            "acp_product": acp,
+        }
+
+
+def register_mcp_tools(mcp: FastAPIMCPServer, agent: BaseRetailAgent) -> None:
+    """Expose MCP tools for ACP transformation workflows."""
+    adapters = getattr(agent, "adapters", build_acp_transformation_adapters())
+
+    async def transform_product(payload: dict[str, Any]) -> dict[str, Any]:
+        sku = payload.get("sku")
+        if not sku:
+            return {"error": "sku is required"}
+        product = await adapters.products.get_product(str(sku))
+        if not product:
+            return {"error": "sku not found", "sku": sku}
+        acp = adapters.mapper.to_acp_product(
+            product,
+            availability=str(payload.get("availability", "in_stock")),
+            currency=str(payload.get("currency", "usd")),
+        )
+        return {"acp_product": acp}
+
+    async def get_product(payload: dict[str, Any]) -> dict[str, Any]:
+        sku = payload.get("sku")
+        if not sku:
+            return {"error": "sku is required"}
+        product = await adapters.products.get_product(str(sku))
+        return {"product": product.model_dump() if product else None}
+
+    mcp.add_tool("/product/acp/transform", transform_product)
+    mcp.add_tool("/product/acp/product", get_product)
+
+
+def _acp_instructions() -> str:
+    return (
+        "You are a product ACP transformation agent. "
+        "Ensure ACP fields are populated, highlight missing inputs, "
+        "and call out any normalization assumptions."
+    )

apps/product-management-acp-transformation/src/product_management_acp_transformation/main.py

@@ -1,14 +1,18 @@
 """ACP transformation 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 product_management_acp_transformation.agents import (
+	ProductAcpTransformationAgent,
+	register_mcp_tools,
+)
+
 SERVICE_NAME = "product-management-acp-transformation"
 memory_settings = MemorySettings()
 app = build_service_app(
 	SERVICE_NAME,
-	agent_class=ServiceAgent,
+	agent_class=ProductAcpTransformationAgent,
 	hot_memory=HotMemory(memory_settings.redis_url),
 	warm_memory=WarmMemory(
 		memory_settings.cosmos_account_uri,
@@ -19,4 +23,5 @@
 		memory_settings.blob_account_url,
 		memory_settings.blob_container,
 	),
+	mcp_setup=register_mcp_tools,
 )

apps/product-management-assortment-optimization/src/product_management_assortment_optimization/adapters.py

@@ -0,0 +1,55 @@
+"""Adapters for the assortment optimization service."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Optional
+
+from holiday_peak_lib.adapters.mock_adapters import MockProductAdapter
+from holiday_peak_lib.adapters.product_adapter import ProductConnector
+from holiday_peak_lib.schemas.product import CatalogProduct
+
+
+@dataclass
+class AssortmentAdapters:
+    """Container for assortment optimization adapters."""
+
+    products: ProductConnector
+    optimizer: "AssortmentOptimizer"
+
+
+class AssortmentOptimizer:
+    """Score products for assortment decisions."""
+
+    async def score_products(self, products: list[CatalogProduct]) -> list[dict[str, Any]]:
+        scored = []
+        for product in products:
+            rating = product.rating if product.rating is not None else 3.5
+            price = product.price if product.price is not None else 0.0
+            score = (rating * 2) - (price / 100)
+            scored.append(
+                {
+                    "sku": product.sku,
+                    "name": product.name,
+                    "rating": rating,
+                    "price": price,
+                    "score": round(score, 3),
+                }
+            )
+        return sorted(scored, key=lambda item: item["score"], reverse=True)
+
+    async def recommend_assortment(
+        self, products: list[CatalogProduct], *, target_size: int
+    ) -> dict[str, Any]:
+        scored = await self.score_products(products)
+        keep = scored[:target_size]
+        drop = scored[target_size:]
+        return {"keep": keep, "drop": drop, "target_size": target_size}
+
+
+def build_assortment_adapters(
+    *, product_connector: Optional[ProductConnector] = None
+) -> AssortmentAdapters:
+    """Create adapters for assortment optimization workflows."""
+    products = product_connector or ProductConnector(adapter=MockProductAdapter())
+    optimizer = AssortmentOptimizer()
+    return AssortmentAdapters(products=products, optimizer=optimizer)

apps/product-management-assortment-optimization/src/product_management_assortment_optimization/agents.py

@@ -0,0 +1,95 @@
+"""Assortment optimization 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 AssortmentAdapters, build_assortment_adapters
+
+
+class AssortmentOptimizationAgent(BaseRetailAgent):
+    """Agent that ranks products for assortment decisions."""
+
+    def __init__(self, config, *args: Any, **kwargs: Any) -> None:
+        super().__init__(config, *args, **kwargs)
+        self._adapters = build_assortment_adapters()
+
+    @property
+    def adapters(self) -> AssortmentAdapters:
+        return self._adapters
+
+    async def handle(self, request: dict[str, Any]) -> dict[str, Any]:
+        skus = [str(sku) for sku in request.get("skus", [])]
+        target_size = int(request.get("target_size", 5))
+        if not skus:
+            return {"error": "skus is required"}
+
+        products = []
+        for sku in skus:
+            product = await self.adapters.products.get_product(sku)
+            if product:
+                products.append(product)
+
+        if not products:
+            return {"error": "no products found", "skus": skus}
+
+        recommendations = await self.adapters.optimizer.recommend_assortment(
+            products, target_size=target_size
+        )
+
+        if self.slm or self.llm:
+            messages = [
+                {"role": "system", "content": _assortment_instructions()},
+                {
+                    "role": "user",
+                    "content": {
+                        "skus": skus,
+                        "products": [p.model_dump() for p in products],
+                        "assortment": recommendations,
+                    },
+                },
+            ]
+            return await self.invoke_model(request=request, messages=messages)
+
+        return {
+            "service": self.service_name,
+            "skus": skus,
+            "assortment": recommendations,
+        }
+
+
+def register_mcp_tools(mcp: FastAPIMCPServer, agent: BaseRetailAgent) -> None:
+    """Expose MCP tools for assortment optimization workflows."""
+    adapters = getattr(agent, "adapters", build_assortment_adapters())
+
+    async def score_products(payload: dict[str, Any]) -> dict[str, Any]:
+        skus = [str(sku) for sku in payload.get("skus", [])]
+        if not skus:
+            return {"error": "skus is required"}
+        products = [p for p in [await adapters.products.get_product(sku) for sku in skus] if p]
+        scored = await adapters.optimizer.score_products(products)
+        return {"scores": scored}
+
+    async def recommend_assortment(payload: dict[str, Any]) -> dict[str, Any]:
+        skus = [str(sku) for sku in payload.get("skus", [])]
+        if not skus:
+            return {"error": "skus is required"}
+        products = [p for p in [await adapters.products.get_product(sku) for sku in skus] if p]
+        target_size = int(payload.get("target_size", 5))
+        recommendations = await adapters.optimizer.recommend_assortment(
+            products, target_size=target_size
+        )
+        return {"assortment": recommendations}
+
+    mcp.add_tool("/assortment/score", score_products)
+    mcp.add_tool("/assortment/recommendations", recommend_assortment)
+
+
+def _assortment_instructions() -> str:
+    return (
+        "You are an assortment optimization agent. "
+        "Rank products by performance indicators and recommend the ideal set. "
+        "Explain trade-offs and highlight missing signals."
+    )

apps/product-management-assortment-optimization/src/product_management_assortment_optimization/main.py

@@ -1,14 +1,18 @@
 """Assortment optimization 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 product_management_assortment_optimization.agents import (
+	AssortmentOptimizationAgent,
+	register_mcp_tools,
+)
+
 SERVICE_NAME = "product-management-assortment-optimization"
 memory_settings = MemorySettings()
 app = build_service_app(
 	SERVICE_NAME,
-	agent_class=ServiceAgent,
+	agent_class=AssortmentOptimizationAgent,
 	hot_memory=HotMemory(memory_settings.redis_url),
 	warm_memory=WarmMemory(
 		memory_settings.cosmos_account_uri,
@@ -19,4 +23,5 @@
 		memory_settings.blob_account_url,
 		memory_settings.blob_container,
 	),
+	mcp_setup=register_mcp_tools,
 )