feat: Mixpeek contextual enrichment for IAB taxonomy classification (#78)

* feat: add Mixpeek contextual enrichment for IAB taxonomy classification

Add MixpeekClient and contextual enrichment tools that enable buyer
agents to classify content into IAB v3.0 taxonomy categories and
search indexed inventory via Mixpeek retriever pipelines.

New files:
- MixpeekClient: async HTTP client for Mixpeek content-intelligence API
- ClassifyContentTool: CrewAI tool for IAB taxonomy classification
- ContextualSearchTool: CrewAI tool for multimodal inventory search
- MCP tools: classify_content and contextual_search via @mcp.tool()
- Unit tests: 18 tests covering client and tool behavior

Configuration: MIXPEEK_API_KEY, MIXPEEK_BASE_URL, MIXPEEK_NAMESPACE
env vars (all optional, tools gracefully degrade when unconfigured).

* feat: add brand-safety scoring, production e2e tests, crew integration

Major improvements to the Mixpeek contextual enrichment integration:

- Rearchitect classify_content to use retriever-based IAB classification
  (semantic search against IAB category corpus) instead of batch taxonomy
  execute endpoint which returns empty for real-time queries
- Add check_brand_safety method and BrandSafetyTool that flags sensitive
  IAB categories (gambling, adult, etc.) with risk levels
- Add auto-discovery of IAB retrievers when no retriever_id is specified
- Wire ClassifyContentTool, BrandSafetyTool, ContextualSearchTool into
  the research crew (channel_crews.py) — tools activate when
  MIXPEEK_API_KEY is configured
- Register check_brand_safety as MCP tool alongside classify_content
  and contextual_search
- Add 16 e2e tests hitting production Mixpeek API (golden_adtech_iab
  namespace with 700+ IAB category documents) verifying:
  - Sports content → "American Football" (score > 0.80)
  - Automotive content → "Automotive" hierarchy (score > 0.80)
  - Food content → "Food & Drink" / "Cooking"
  - Tech content → "Artificial Intelligence" (score > 0.85)
  - Gambling content flagged as brand-unsafe (high risk)
  - Safe content not flagged
  - Threshold filtering works correctly
  - Error handling for invalid keys and retriever IDs

All 41 tests pass (25 unit + 16 e2e against production).

---------

Co-authored-by: Siraj M <sirajkodescan@gmail.com>

Author: esteininger

.env.example

@@ -40,6 +40,11 @@ POSTGRES_POOL_MAX=10
 # Routes durable data (deals, campaigns, orders) to Postgres
 # and ephemeral data (sessions, caches, locks) to Redis
 
+# Mixpeek Contextual Enrichment (optional: enables classify_content and contextual_search tools)
+MIXPEEK_API_KEY=
+MIXPEEK_BASE_URL=https://api.mixpeek.com
+MIXPEEK_NAMESPACE=
+
 # Environment
 ENVIRONMENT=development
 LOG_LEVEL=INFO

src/ad_buyer/clients/__init__.py

@@ -6,6 +6,7 @@
 from .a2a_client import A2AClient, A2AError, A2AResponse
 from .deals_client import DealsClient, DealsClientError
 from .mcp_client import IABMCPClient, MCPClientError, MCPToolResult
+from .mixpeek_client import MixpeekClient, MixpeekError
 from .opendirect_client import OpenDirectClient
 from .sgp_client import SGPAuthError, SGPClient, SGPClientError
 from .ucp_client import UCPClient, UCPExchangeResult
@@ -32,6 +33,9 @@
     # IAB Deals API v1.0 client (quote-then-book flow)
     "DealsClient",
     "DealsClientError",
+    # Mixpeek contextual enrichment
+    "MixpeekClient",
+    "MixpeekError",
     # IAB Diligence Platform (SGP) approval gate
     "SGPClient",
     "SGPClientError",

src/ad_buyer/clients/mixpeek_client.py

@@ -0,0 +1,294 @@
+# Mixpeek contextual enrichment client for the Ad Buyer Agent.
+#
+# Provides IAB taxonomy classification via retriever pipelines,
+# brand-safety scoring, and contextual inventory search via the
+# Mixpeek REST API.
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+# Default timeout for Mixpeek API calls (seconds).
+_DEFAULT_TIMEOUT = 30.0
+
+# Brand-safety sensitive IAB categories that advertisers typically
+# want to avoid or require explicit opt-in for.
+BRAND_UNSAFE_CATEGORIES = frozenset({
+    "Poker and Professional Gambling",
+    "Casinos & Gambling",
+    "Casino Games",
+    "Lotteries and Scratchcards",
+    "Sensitive Topics",
+    "Adult Content",
+    "Illegal Content",
+    "Debated Sensitive Social Topics",
+    "Terrorism",
+    "Crime",
+    "Drugs",
+    "Tobacco",
+    "Arms & Ammunition",
+    "Death & Grieving",
+})
+
+
+class MixpeekError(Exception):
+    """Raised when a Mixpeek API call fails."""
+
+    def __init__(self, message: str, status_code: int | None = None):
+        super().__init__(message)
+        self.status_code = status_code
+
+
+class MixpeekClient:
+    """Async HTTP client for the Mixpeek content-intelligence API.
+
+    The client wraps capabilities useful during buyer-agent research
+    and execution phases:
+
+    1. **IAB classification** – classify text/URL content against IAB
+       v3.0 taxonomy categories using a retriever pipeline that performs
+       semantic search against an IAB category reference corpus.
+    2. **Brand-safety check** – score content for brand-safety risk
+       by identifying sensitive IAB categories in the classification.
+    3. **Contextual search** – search indexed ad inventory via retriever
+       pipelines combining multimodal search, taxonomy enrichment,
+       and reranking.
+
+    All methods are async and use ``httpx.AsyncClient`` under the hood.
+    """
+
+    def __init__(
+        self,
+        api_key: str,
+        base_url: str = "https://api.mixpeek.com",
+        namespace: str | None = None,
+        timeout: float = _DEFAULT_TIMEOUT,
+    ):
+        self.api_key = api_key
+        self.base_url = base_url.rstrip("/")
+        self.namespace = namespace
+        self._client = httpx.AsyncClient(
+            base_url=self.base_url,
+            timeout=timeout,
+            follow_redirects=True,
+        )
+
+    # ------------------------------------------------------------------
+    # Internal helpers
+    # ------------------------------------------------------------------
+
+    def _headers(self, namespace: str | None = None) -> dict[str, str]:
+        """Build request headers with auth and optional namespace."""
+        h: dict[str, str] = {
+            "Authorization": f"Bearer {self.api_key}",
+            "Content-Type": "application/json",
+        }
+        ns = namespace or self.namespace
+        if ns:
+            h["X-Namespace"] = ns
+        return h
+
+    async def _request(
+        self,
+        method: str,
+        path: str,
+        *,
+        namespace: str | None = None,
+        json: dict | None = None,
+        params: dict | None = None,
+    ) -> dict[str, Any]:
+        """Send a request and return the parsed JSON response."""
+        try:
+            resp = await self._client.request(
+                method,
+                path,
+                headers=self._headers(namespace),
+                json=json,
+                params=params,
+            )
+        except httpx.HTTPError as exc:
+            raise MixpeekError(f"HTTP error: {exc}") from exc
+
+        if resp.status_code >= 400:
+            raise MixpeekError(
+                f"Mixpeek API error {resp.status_code}: {resp.text}",
+                status_code=resp.status_code,
+            )
+        return resp.json()
+
+    # ------------------------------------------------------------------
+    # Public API
+    # ------------------------------------------------------------------
+
+    async def list_taxonomies(self, namespace: str | None = None) -> list[dict]:
+        """List available taxonomies in the namespace."""
+        data = await self._request(
+            "POST", "/v1/taxonomies/list", namespace=namespace, json={}
+        )
+        return data.get("results", [])
+
+    async def list_retrievers(self, namespace: str | None = None) -> list[dict]:
+        """List available retriever pipelines in the namespace."""
+        data = await self._request(
+            "POST", "/v1/retrievers/list", namespace=namespace, json={}
+        )
+        return data.get("results", [])
+
+    async def classify_content(
+        self,
+        retriever_id: str,
+        text: str,
+        *,
+        namespace: str | None = None,
+        limit: int = 10,
+    ) -> dict[str, Any]:
+        """Classify content into IAB taxonomy categories.
+
+        Uses a retriever pipeline that performs semantic search against
+        an IAB category reference corpus.  Each result represents an
+        IAB category match with a confidence score.
+
+        Args:
+            retriever_id: Retriever pipeline configured for IAB search.
+            text: Content text to classify.
+            namespace: Override namespace for this call.
+            limit: Max category matches to return.
+
+        Returns:
+            Dict with ``documents`` list, each containing
+            ``iab_category_name``, ``iab_path``, ``iab_tier``,
+            and ``score``.
+        """
+        body: dict[str, Any] = {
+            "inputs": {"query": text},
+            "page_size": limit,
+        }
+        return await self._request(
+            "POST",
+            f"/v1/retrievers/{retriever_id}/execute",
+            namespace=namespace,
+            json=body,
+        )
+
+    async def check_brand_safety(
+        self,
+        retriever_id: str,
+        text: str,
+        *,
+        namespace: str | None = None,
+        threshold: float = 0.80,
+        limit: int = 10,
+    ) -> dict[str, Any]:
+        """Check content for brand-safety risk.
+
+        Classifies content via the IAB retriever and flags any matches
+        against known sensitive categories (gambling, adult, etc.).
+
+        Args:
+            retriever_id: Retriever pipeline configured for IAB search.
+            text: Content text to evaluate.
+            namespace: Override namespace for this call.
+            threshold: Minimum score to consider a category match.
+            limit: Max category matches to evaluate.
+
+        Returns:
+            Dict with ``safe`` bool, ``risk_level`` (low/medium/high),
+            ``flagged_categories`` list, and full ``categories`` list.
+        """
+        result = await self.classify_content(
+            retriever_id=retriever_id,
+            text=text,
+            namespace=namespace,
+            limit=limit,
+        )
+
+        docs = result.get("documents", [])
+        categories = []
+        flagged = []
+
+        for doc in docs:
+            score = doc.get("score", 0)
+            if score < threshold:
+                continue
+            cat_name = doc.get("iab_category_name", "")
+            entry = {
+                "category": cat_name,
+                "path": doc.get("iab_path", []),
+                "tier": doc.get("iab_tier"),
+                "score": score,
+            }
+            categories.append(entry)
+            if cat_name in BRAND_UNSAFE_CATEGORIES:
+                flagged.append(entry)
+
+        if flagged:
+            max_score = max(f["score"] for f in flagged)
+            risk_level = "high" if max_score >= 0.85 else "medium"
+        else:
+            risk_level = "low"
+
+        return {
+            "safe": len(flagged) == 0,
+            "risk_level": risk_level,
+            "flagged_categories": flagged,
+            "categories": categories,
+        }
+
+    async def search_content(
+        self,
+        retriever_id: str,
+        query: str,
+        *,
+        namespace: str | None = None,
+        limit: int = 10,
+        filters: dict | None = None,
+    ) -> dict[str, Any]:
+        """Execute a retriever pipeline (contextual search).
+
+        The retriever can combine feature search, brand-safety filtering,
+        taxonomy enrichment, and reranking in a single call.
+        """
+        inputs: dict[str, Any] = {"query": query}
+        if filters:
+            inputs.update(filters)
+
+        body: dict[str, Any] = {
+            "inputs": inputs,
+            "page_size": limit,
+        }
+
+        return await self._request(
+            "POST",
+            f"/v1/retrievers/{retriever_id}/execute",
+            namespace=namespace,
+            json=body,
+        )
+
+    async def get_tools(self) -> list[dict]:
+        """Fetch the public MCP tools list (no auth required).
+
+        Hits the ``/tools`` REST endpoint on the MCP server, which
+        returns all 48 tool definitions without authentication.
+        """
+        resp = await self._client.get(
+            "https://mcp.mixpeek.com/tools",
+            timeout=10,
+        )
+        return resp.json().get("tools", [])
+
+    async def health(self) -> dict[str, Any]:
+        """Check MCP server health (no auth required)."""
+        resp = await self._client.get(
+            "https://mcp.mixpeek.com/health",
+            timeout=5,
+        )
+        return resp.json()
+
+    async def close(self) -> None:
+        """Shut down the underlying HTTP client."""
+        await self._client.aclose()

src/ad_buyer/config/settings.py

@@ -93,6 +93,11 @@ def get_cors_origins(self) -> list[str]:
             return []
         return [o.strip() for o in self.cors_allowed_origins.split(",") if o.strip()]
 
+    # Mixpeek contextual enrichment
+    mixpeek_api_key: str = ""
+    mixpeek_base_url: str = "https://api.mixpeek.com"
+    mixpeek_namespace: str = ""
+
     # Environment
     environment: str = "development"
     log_level: str = "INFO"

src/ad_buyer/crews/channel_crews.py

@@ -44,16 +44,31 @@
 from ..tools.execution.line_management import BookLineTool, CreateLineTool, ReserveLineTool
 from ..tools.execution.order_management import CreateOrderTool
 from ..tools.research.avails_check import AvailsCheckTool
+from ..tools.research.contextual_enrichment import (
+    BrandSafetyTool,
+    ClassifyContentTool,
+    ContextualSearchTool,
+)
 from ..tools.research.product_search import ProductSearchTool
 
 
 def _create_research_tools(client: OpenDirectClient) -> list[Any]:
     """Create research tools with the OpenDirect client."""
-    return [
+    tools: list[Any] = [
         ProductSearchTool(client),
         AvailsCheckTool(client),
     ]
 
+    # Add Mixpeek contextual enrichment tools when configured
+    if settings.mixpeek_api_key:
+        tools.extend([
+            ClassifyContentTool(),
+            BrandSafetyTool(),
+            ContextualSearchTool(),
+        ])
+
+    return tools
+
 
 def _create_execution_tools(client: OpenDirectClient) -> list[Any]:
     """Create execution tools with the OpenDirect client."""