diff --git a/docs/integration/deals-api-mcp.md b/docs/integration/deals-api-mcp.md new file mode 100644 index 0000000..d127613 --- /dev/null +++ b/docs/integration/deals-api-mcp.md @@ -0,0 +1,141 @@ +# deals-api-mcp Integration + +The seller agent can distribute deals through [deals-api-mcp](https://github.com/IABTechLab/deals-api-mcp) — an MCP server that implements the IAB Deal Sync API v1.0. The seller agent talks to it over **MCP Streamable HTTP**. + +## How It Fits Together + +``` +seller-agent (Python) + └── DealsAPIMCPClient ← DealSyncClient implementation + └── MCP Transport ← MCP Streamable HTTP transport + └── deals-api-mcp ← MCP server (TypeScript, SQLite) +``` + +`DealsAPIMCPClient` implements the `DealSyncClient` interface and is registered in the deal-sync registry (a peer of the SSP registry). All operations are exposed through the seller agent's standard REST API — no direct access to the connector is needed. + +## Prerequisites + +Start `deals-api-mcp` in HTTP mode before the seller agent connects: + +```bash +MCP_TRANSPORT=http MCP_PORT=3100 NODE_ENV=demo node dist/index.js +``` + +The server listens at `http://localhost:3100/mcp`. See [deals-api-mcp HTTP Streamable Transport](https://github.com/IABTechLab/deals-api-mcp#http-streamable-transport) for full env var details. + +## Configuration + +Enable the connector via environment variables: + +```bash +DEAL_SYNC_CONNECTORS=deals_api_mcp +DEALS_API_MCP_URL=http://localhost:3100/mcp +DEALS_API_MCP_SELLER_ORIGIN=publisher.example.com +DEALS_API_MCP_KEY= # omit for NODE_ENV=demo +``` + +| Variable | Description | +|----------|-------------| +| `DEAL_SYNC_CONNECTORS` | Comma-separated list of active deal-sync providers. Include `deals_api_mcp` to enable. | +| `DEALS_API_MCP_URL` | Full URL of the deals-api-mcp `/mcp` endpoint | +| `DEALS_API_MCP_SELLER_ORIGIN` | Publisher domain stamped on every created deal | +| `DEALS_API_MCP_KEY` | Auth key. Required when `NODE_ENV=production`; omit in demo mode | + +## Operations + +### Distribute (Create) a Deal + +```bash +curl -X POST http://localhost:8000/api/v1/deals/distribute \ + -H "Authorization: Bearer " \ + -H "Content-Type: application/json" \ + -d '{ + "deal_id": "your-internal-deal-id", + "ssp_name": "deals_api_mcp", + "name": "Q3 Sports Video", + "advertiser": "Acme Corp", + "cpm": 25.00, + "deal_type": "PMP", + "start_date": "2026-07-01T00:00:00Z", + "end_date": "2026-09-30T00:00:00Z", + "buyer_seat_ids": ["seat-001"] + }' +``` + +Response: + +```json +{ + "deal_id": "8b3f9cfe-7716-48e0-94bf-42fe44595928", + "external_deal_id": "IAB-Q3SPORTS001", + "channel": "deal_sync", + "provider": "deals_api_mcp", + "provider_name": "IAB Deals MCP", + "status": "created", + "deal": { "name": "Q3 Sports Video", "cpm": 25.00, "external_deal_id": "IAB-Q3SPORTS001", ... } +} +``` + +`deal_id` is the internal UUID (e.g. `8b3f9cfe-...`) required by MCP tools. `external_deal_id` is the OpenRTB / IAB deal ID (e.g. `"IAB-..."`) that DSPs need for activation — it is also present on the nested `deal` object and in `DEAL_SYNCED` event payloads. + +Omit `ssp_name` to let the seller agent route based on `SSP_ROUTING_RULES`. + +### Troubleshoot a Deal + +```bash +curl "http://localhost:8000/api/v1/deals/8b3f9cfe-7716-48e0-94bf-42fe44595928/ssp-troubleshoot?ssp_name=deals_api_mcp" \ + -H "Authorization: Bearer " +``` + +Returns the deal's buyer seat statuses and any rejected seats flagged as primary issues. + +## Status Mapping + +`deals-api-mcp` returns seller status as an integer (`sellerStatus`). The connector normalizes it: + +| `sellerStatus` | IAB label | Normalized status | +|---------------|-----------|-------------------| +| `0` | Active | `active` | +| `1` | Paused | `paused` | +| `2` | Pending | `created` | +| `4` | Complete | `expired` | +| `5` | Archived | `archived` | + +!!! warning "Buyer vs seller status codes share the integer space" + A buyer-side `0` (Pending) is not the same as a deal-side `0` (Active). They live on different objects — `deal.sellerStatus` vs `buyerSeat.buyerStatus`. Always check which field you are reading. + +## Connection Lifecycle + +```mermaid +sequenceDiagram + participant SA as seller-agent + participant TR as MCP Transport + participant MCP as deals-api-mcp + + SA->>TR: __aenter__ (first request) + TR->>MCP: POST /mcp initialize + MCP-->>TR: 200 OK + Mcp-Session-Id: + TR-->>SA: persistent session ready ✓ + + SA->>TR: call_tool("deals_create", args) + TR->>MCP: POST /mcp tools/call (Mcp-Session-Id: ) + MCP-->>TR: Content-Type: text/event-stream\ndata: {"result":{...}} + TR-->>SA: parsed result dict + + SA->>TR: __aenter__ (second request — session reused) + Note over SA,TR: background task still running, no re-initialize + SA->>TR: call_tool("deals_create", args) + TR->>MCP: POST /mcp tools/call (Mcp-Session-Id: ) + MCP-->>TR: Content-Type: text/event-stream\ndata: {"result":{...}} + TR-->>SA: parsed result dict +``` + +!!! note "Persistent session" + `DealsAPIMCPClient` keeps a **class-level persistent background task** so the MCP session is created on first use and reused across subsequent `async with` blocks without re-initializing. + + If the connection drops, the stale session is detected on the next tool call (via a 30 s timeout), the background task is restarted, and the call is retried automatically — no process restart required. The first request after a drop may take up to 30 seconds while reconnection completes; subsequent requests are unaffected. + +## Related + +- [deals-api-mcp README](https://github.com/IABTechLab/deals-api-mcp#readme) — server setup, all 10 MCP tools, data models +- `src/ad_seller/clients/deals_api_mcp_client.py` — source for `DealsAPIMCPClient` diff --git a/mkdocs.yml b/mkdocs.yml index c149714..8bfbba0 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -76,6 +76,7 @@ nav: - Integration: - Buyer Agent Guide: integration/buyer-agent.md - Negotiation Protocol: integration/negotiation.md + - deals-api-mcp: integration/deals-api-mcp.md - AI Assistant Setup: - Claude (Desktop & Web): guides/claude-desktop-setup.md - ChatGPT, Codex & AI IDEs: guides/chatgpt-setup.md diff --git a/src/ad_seller/clients/__init__.py b/src/ad_seller/clients/__init__.py index 5698969..cd3e219 100644 --- a/src/ad_seller/clients/__init__.py +++ b/src/ad_seller/clients/__init__.py @@ -1,7 +1,7 @@ # Author: Green Mountain Systems AI Inc. # Donated to IAB Tech Lab -"""Protocol clients for OpenDirect 2.1, A2A, GAM, SSP, and ad server abstraction.""" +"""Protocol clients for OpenDirect 2.1, A2A, GAM, SSP, deal-sync, and ad server abstraction.""" from .a2a_client import A2AClient, A2AResponse from .ad_server_base import ( @@ -16,6 +16,9 @@ get_ad_server_client, ) from .csv_adapter import CSVAdServerClient +from .deal_sync_base import DealSyncClient, DealSyncProvider, DealSyncRegistry +from .deal_sync_factory import build_deal_sync_registry +from .deals_api_mcp_client import DealsAPIMCPClient from .freewheel_adapter import FreeWheelAdServerClient from .gam_rest_client import GAMRestClient from .gam_soap_client import GAMSoapClient @@ -65,4 +68,10 @@ "RESTSSPClient", "build_ssp_registry", "IndexExchangeSSPClient", + # Deal-sync abstraction + "DealSyncClient", + "DealSyncProvider", + "DealSyncRegistry", + "DealsAPIMCPClient", + "build_deal_sync_registry", ] diff --git a/src/ad_seller/clients/deal_sync_base.py b/src/ad_seller/clients/deal_sync_base.py new file mode 100644 index 0000000..2b648aa --- /dev/null +++ b/src/ad_seller/clients/deal_sync_base.py @@ -0,0 +1,120 @@ +# Author: Green Mountain Systems AI Inc. +# Donated to IAB Tech Lab + +"""Abstract deal-sync client interface. + +Deal-sync connectors push negotiated deals to an external deal +synchronization service (e.g. the IAB deals-api-mcp server), which +propagates them to buyer-side providers. This is a peer of the other +connector families: + - AdServerClient: inventory sync, deal setup in the publisher's ad server + - SSPClient: deal distribution through SSP exchanges to DSPs + - DealSyncClient: deal sync through an external deal-sync service + +Reuses the normalized deal models from ssp_base (SSPDeal, +SSPDealCreateRequest, SSPDealStatus); giving this family its own +models is out of scope for the connector split. +""" + +from abc import ABC, abstractmethod +from enum import Enum +from typing import Any, ClassVar, Optional + +from .ssp_base import SSPDeal, SSPDealCreateRequest, SSPDealStatus + + +class DealSyncProvider(str, Enum): + """Known deal-sync providers (extensible via config).""" + + DEALS_API_MCP = "deals_api_mcp" + + +class DealSyncClient(ABC): + """Abstract base class for deal-sync integrations. + + Each provider implementation must provide these methods. + The deal-sync registry manages configured providers. + """ + + channel: ClassVar[str] = "deal_sync" + provider: DealSyncProvider + provider_name: str = "Unknown Deal Sync Provider" + + @abstractmethod + async def connect(self) -> None: + """Establish connection to the deal-sync service.""" + ... + + @abstractmethod + async def disconnect(self) -> None: + """Disconnect from the deal-sync service.""" + ... + + async def __aenter__(self) -> "DealSyncClient": + await self.connect() + return self + + async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: + await self.disconnect() + + @abstractmethod + async def create_deal(self, request: SSPDealCreateRequest) -> SSPDeal: + """Create a new deal on the deal-sync service.""" + ... + + @abstractmethod + async def get_deal(self, deal_id: str) -> SSPDeal: + """Get deal details (and sync status) by provider-internal ID.""" + ... + + @abstractmethod + async def list_deals( + self, + *, + status: Optional[SSPDealStatus] = None, + limit: int = 100, + ) -> list[SSPDeal]: + """List deals on this provider.""" + ... + + @abstractmethod + async def update_deal(self, deal_id: str, updates: dict[str, Any]) -> SSPDeal: + """Update mutable deal attributes.""" + ... + + async def health_check(self) -> bool: + """Check if the connection is healthy. Override for custom logic.""" + return True + + +class DealSyncRegistry: + """Registry for configured deal-sync clients, keyed by provider name.""" + + def __init__(self) -> None: + self._clients: dict[str, DealSyncClient] = {} + self._default: Optional[str] = None + + def register(self, name: str, client: DealSyncClient) -> None: + """Register a deal-sync client by provider name.""" + self._clients[name] = client + if self._default is None: + self._default = name + + def get_client(self, name: str) -> DealSyncClient: + """Get a deal-sync client by provider name.""" + if name not in self._clients: + raise KeyError( + f"Deal-sync provider '{name}' not registered. " + f"Available: {list(self._clients.keys())}" + ) + return self._clients[name] + + def get_default(self) -> DealSyncClient: + """Get the default (first-registered) deal-sync client.""" + if not self._default: + raise RuntimeError("No deal-sync clients registered") + return self._clients[self._default] + + def list_providers(self) -> list[str]: + """List registered provider names.""" + return list(self._clients.keys()) diff --git a/src/ad_seller/clients/deal_sync_factory.py b/src/ad_seller/clients/deal_sync_factory.py new file mode 100644 index 0000000..934a4f9 --- /dev/null +++ b/src/ad_seller/clients/deal_sync_factory.py @@ -0,0 +1,54 @@ +# Author: Green Mountain Systems AI Inc. +# Donated to IAB Tech Lab + +"""Deal-sync registry factory — builds deal-sync clients from config. + +Reads DEAL_SYNC_CONNECTORS from settings to build a DealSyncRegistry +with all configured providers. +""" + +import logging +from typing import Any + +from .deal_sync_base import DealSyncRegistry + +logger = logging.getLogger(__name__) + + +def build_deal_sync_registry(settings: Any = None) -> DealSyncRegistry: + """Build a DealSyncRegistry from application settings. + + Reads DEAL_SYNC_CONNECTORS (comma-separated list) and creates the + appropriate client for each configured provider. + """ + if settings is None: + from ..config import get_settings + + settings = get_settings() + + registry = DealSyncRegistry() + + connectors = [s.strip() for s in settings.deal_sync_connectors.split(",") if s.strip()] + + for name in connectors: + name_lower = name.lower() + if name_lower == "deals_api_mcp": + from .deals_api_mcp_client import DealsAPIMCPClient + + if not settings.deals_api_mcp_url: + logger.warning("deals_api_mcp configured but DEALS_API_MCP_URL not set") + continue + + registry.register( + name_lower, + DealsAPIMCPClient( + mcp_url=settings.deals_api_mcp_url, + api_key=settings.deals_api_mcp_key, + seller_origin=settings.deals_api_mcp_seller_origin, + ), + ) + logger.info("Registered deal-sync connector: %s", name_lower) + else: + logger.warning("Unknown deal-sync provider '%s'", name) + + return registry diff --git a/src/ad_seller/clients/deals_api_mcp_client.py b/src/ad_seller/clients/deals_api_mcp_client.py new file mode 100644 index 0000000..a9637b1 --- /dev/null +++ b/src/ad_seller/clients/deals_api_mcp_client.py @@ -0,0 +1,424 @@ +# Author: Green Mountain Systems AI Inc. +# Donated to IAB Tech Lab + +"""Deal-sync connector for IAB deals-api-mcp (HTTP Streamable MCP transport). + +Connects to a running deals-api-mcp server via MCP Streamable HTTP and maps +the IAB Deal Sync API v1.0 tool schema to the DealSyncClient interface. + +deals-api-mcp tools used: + - deals_create: create a new deal (required: name, origin, seller, dealFloor, startDate) + - deals_status: get deal + all buyer seat statuses + history + - deals_list: list deals with optional status filter + - deals_update: update mutable deal fields (blocked after deals_send) + - deals_pause: pause an active deal (propagates to provider) + - deals_resume: resume a paused deal (propagates to provider) + +Note on deal IDs: + SSPDeal.deal_id holds the internal UUID returned by deals-api-mcp (deal.id). + All MCP tools validate dealId as z.string().uuid() — they require this UUID. + SSPDeal.external_deal_id holds the OpenRTB / IAB deal ID (externalDealId, + e.g. "IAB-...") used for DSP activation. +""" + +import asyncio +import logging +from datetime import datetime, timezone +from typing import Any, ClassVar, Optional + +from .deal_sync_base import DealSyncClient, DealSyncProvider +from .freewheel_mcp_client import FreeWheelMCPClient +from .ssp_base import ( + SSPDeal, + SSPDealCreateRequest, + SSPDealStatus, + SSPDealType, + SSPTroubleshootResult, +) + +logger = logging.getLogger(__name__) + + +class _StaleSessionError(Exception): + """Internal sentinel raised when a tool call signals the session is gone.""" + +# deals-api-mcp sellerStatus integer → SSPDealStatus +# SellerStatus enum: 0=Active, 1=Paused, 2=Pending, 4=Complete, 5=Archived +_SELLER_STATUS_MAP: dict[int, SSPDealStatus] = { + 0: SSPDealStatus.ACTIVE, + 1: SSPDealStatus.PAUSED, + 2: SSPDealStatus.CREATED, + 4: SSPDealStatus.EXPIRED, + 5: SSPDealStatus.ARCHIVED, +} + +# Reverse: SSPDealStatus → sellerStatus integer (for deals_list filter) +_STATUS_TO_SELLER_INT: dict[SSPDealStatus, int] = {v: k for k, v in _SELLER_STATUS_MAP.items()} + +_SELLER_STATUS_LABEL: dict[int, str] = { + 0: "Active", + 1: "Paused", + 2: "Pending", + 4: "Complete", + 5: "Archived", +} + + +class DealsAPIMCPClient(DealSyncClient): + """Deal-sync connector for deals-api-mcp via MCP Streamable HTTP. + + Wraps FreeWheelMCPClient for transport and maps structured IAB tool + arguments to/from the DealSyncClient interface. + + A class-level background task holds one shared MCP session open so + successive ``async with`` blocks reuse it without re-initializing. + ``__aexit__`` is a no-op between requests. If the connection drops (or the + URL changes), the next ``__aenter__`` tears down the old task and starts a + fresh session. + """ + + provider = DealSyncProvider.DEALS_API_MCP + provider_name: str = "IAB Deals MCP" + ssp_name: str = "IAB Deals MCP" # feeds SSPDeal.ssp_name in _parse_deal + + # ── Class-level persistent session ───────────────────────────────────── + _shared_mcp: ClassVar[Optional[FreeWheelMCPClient]] = None + _session_task: ClassVar[Optional[asyncio.Task]] = None + _session_ready: ClassVar[Optional[asyncio.Event]] = None + _session_done: ClassVar[Optional[asyncio.Event]] = None + _session_error: ClassVar[Optional[Exception]] = None + _session_url: ClassVar[Optional[str]] = None + _session_lock: ClassVar[Optional[asyncio.Lock]] = None + + @classmethod + def _get_lock(cls) -> asyncio.Lock: + if cls._session_lock is None: + cls._session_lock = asyncio.Lock() + return cls._session_lock + + def __init__( + self, + *, + mcp_url: str, + api_key: Optional[str] = None, + seller_origin: str = "publisher.example.com", + ) -> None: + self.ssp_name = "IAB Deals MCP" + self._mcp_url = mcp_url + self._api_key = api_key + self._seller_origin = seller_origin + self._mcp_client = FreeWheelMCPClient() # replaced with shared client in __aenter__ + + # ── Lifecycle ────────────────────────────────────────────────────────── + + async def connect(self) -> None: + auth_params = {"api_key": self._api_key} if self._api_key else None + await self._mcp_client.connect(url=self._mcp_url, auth_params=auth_params) + logger.info("Connected to deals-api-mcp at %s", self._mcp_url) + + async def disconnect(self) -> None: + await self._mcp_client.disconnect() + + async def __aenter__(self) -> "DealsAPIMCPClient": + """Ensure the persistent class-level MCP session is running, then wire + self._mcp_client to it. Satisfies anyio's cancel-scope invariant by + running the full streamablehttp_client lifecycle inside a single + background asyncio Task. + """ + cls = DealsAPIMCPClient + async with self._get_lock(): + session_alive = ( + cls._session_task is not None + and not cls._session_task.done() + and cls._session_url == self._mcp_url + and cls._shared_mcp is not None + and cls._shared_mcp._connected + ) + if not session_alive: + await self._start_shared_session() + + if cls._session_error: + raise cls._session_error + + self._mcp_client = cls._shared_mcp # type: ignore[assignment] + return self + + async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: + # Keep the shared session open between requests; reconnect happens on + # the next __aenter__ if the background task has died. + pass + + @classmethod + async def _stop_shared_session(cls) -> None: + """Signal the background session to exit and wait for it to finish.""" + done = cls._session_done + task = cls._session_task + if done is not None: + done.set() + if task is not None and not task.done(): + try: + await asyncio.wait_for(task, timeout=5.0) + except (TimeoutError, asyncio.CancelledError, Exception): + task.cancel() + try: + await task + except (asyncio.CancelledError, Exception): + pass + if cls._shared_mcp is not None: + cls._shared_mcp._connected = False + cls._shared_mcp._session = None + cls._session_task = None + cls._session_ready = None + cls._session_done = None + cls._session_error = None + + async def _start_shared_session(self) -> None: + """Start a new background task that holds the shared MCP session open. + + Stops any prior shared session first so a drop (or URL change) can + reconnect without leaking the old transport task. + """ + from mcp import ClientSession + from mcp.client.streamable_http import streamablehttp_client + + cls = DealsAPIMCPClient + await cls._stop_shared_session() + + cls._session_ready = asyncio.Event() + cls._session_done = asyncio.Event() + cls._session_error = None + cls._shared_mcp = FreeWheelMCPClient() + cls._session_url = self._mcp_url + + mcp_url = self._mcp_url + api_key = self._api_key + shared_mcp = cls._shared_mcp + ready = cls._session_ready + done = cls._session_done + + async def _run_session() -> None: + try: + headers = {"x-api-key": api_key} if api_key else None + async with streamablehttp_client(mcp_url, headers=headers) as (read, write, _): + async with ClientSession(read, write) as session: + await session.initialize() + shared_mcp._session = session + shared_mcp._connected = True + ready.set() + logger.info("Persistent MCP session established at %s", mcp_url) + await done.wait() # holds connection open until stop/reconnect + except Exception as exc: + cls._session_error = exc + if not ready.is_set(): + ready.set() + finally: + shared_mcp._connected = False + shared_mcp._session = None + + cls._session_task = asyncio.create_task(_run_session()) + await cls._session_ready.wait() + + # ── Tool call wrapper ────────────────────────────────────────────────── + + async def _call_tool(self, tool_name: str, args: dict[str, Any]) -> Any: + """Call an MCP tool, detecting stale sessions and reconnecting once. + + Two failure modes after a server restart: + - Hang: tool call POST gets 400 but SDK waits for SSE response that + never arrives (session_alive lied — SDK reconnect loop kept the + background task running with the old session ID). A timeout breaks + the hang and triggers reconnect. + - Immediate error: SDK raises httpx.HTTPStatusError(400/404). Caught + here and also triggers reconnect. + """ + import httpx + + _TOOL_TIMEOUT = 30.0 # seconds before treating a call as hung/stale + + async def _attempt() -> Any: + try: + return await asyncio.wait_for( + self._mcp_client.call_tool(tool_name, args), + timeout=_TOOL_TIMEOUT, + ) + except (TimeoutError, asyncio.TimeoutError): + raise _StaleSessionError("tool call timed out — stale session suspected") + except httpx.HTTPStatusError as exc: + if exc.response.status_code in (400, 404): + raise _StaleSessionError(str(exc)) from exc + raise + except Exception as exc: + msg = str(exc).lower() + if "400" in msg or "404" in msg or "bad request" in msg or "not found" in msg: + raise _StaleSessionError(str(exc)) from exc + raise + + try: + return await _attempt() + except _StaleSessionError: + logger.info("Stale MCP session detected — reconnecting to %s", self._mcp_url) + async with self._get_lock(): + await self._start_shared_session() + if DealsAPIMCPClient._session_error: + raise DealsAPIMCPClient._session_error + self._mcp_client = DealsAPIMCPClient._shared_mcp # type: ignore[assignment] + return await _attempt() + + # ── Deal Operations ──────────────────────────────────────────────────── + + async def create_deal(self, request: SSPDealCreateRequest) -> SSPDeal: + """Map SSPDealCreateRequest → deals_create structured args.""" + cpm = getattr(request, "cpm", None) + if cpm is None: + raise ValueError("cpm is required to create a deal via deals-api-mcp") + + now_iso = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + deal_type = getattr(request, "deal_type", None) + + args: dict[str, Any] = { + "name": getattr(request, "name", None) or "Untitled Deal", + "origin": self._seller_origin, + "seller": self._seller_origin, + "dealFloor": cpm, + "startDate": getattr(request, "start_date", None) or now_iso, + } + + # PG deals must be flagged as guaranteed + if deal_type == SSPDealType.PG: + args["guar"] = 1 + + if getattr(request, "end_date", None): + args["endDate"] = request.end_date + if getattr(request, "impressions_goal", None): + args["units"] = request.impressions_goal + if getattr(request, "buyer_seat_ids", None): + args["wseat"] = request.buyer_seat_ids + if getattr(request, "description", None): + args["description"] = request.description + if getattr(request, "currency", None): + args["currency"] = request.currency + + raw = await self._call_tool("deals_create", args) + deal = self._parse_deal(raw) + # deals-api-mcp has no dealType concept — echo back the requested type + # so callers aren't silently told every deal is PMP. + if deal_type: + deal = deal.model_copy(update={"deal_type": deal_type}) + return deal + + async def get_deal(self, deal_id: str) -> SSPDeal: + raw = await self._call_tool("deals_status", {"dealId": deal_id}) + return self._parse_deal(raw) + + async def list_deals( + self, + *, + status: Optional[SSPDealStatus] = None, + limit: int = 100, + ) -> list[SSPDeal]: + args: dict[str, Any] = {"pageSize": min(limit, 100)} + if status is not None and status in _STATUS_TO_SELLER_INT: + args["status"] = _STATUS_TO_SELLER_INT[status] + raw = await self._call_tool("deals_list", args) + if isinstance(raw, dict): + items = raw.get("deals", raw.get("items", [])) + return [self._parse_deal({"deal": d}) for d in items] + return [] + + async def clone_deal( + self, + source_deal_id: str, + overrides: Optional[dict[str, Any]] = None, + ) -> SSPDeal: + """Clone by fetching the source deal and creating a new one with overrides.""" + source_raw = await self._call_tool("deals_status", {"dealId": source_deal_id}) + + # Build create args from source deal's terms, apply overrides + source_deal = source_raw.get("deal", {}) if isinstance(source_raw, dict) else {} + terms = source_deal.get("terms", {}) if isinstance(source_deal.get("terms"), dict) else {} + now_iso = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + overrides = overrides or {} + + # Never invent a floor — same rule as create_deal. + floor = overrides.get("dealFloor", terms.get("dealFloor")) + if floor is None: + raise ValueError("dealFloor is required to clone a deal via deals-api-mcp") + + args: dict[str, Any] = { + "name": f"Copy of {source_deal.get('name', source_deal_id)}", + "origin": self._seller_origin, + "seller": source_deal.get("seller", self.ssp_name), + "dealFloor": floor, + "startDate": terms.get("startDate", now_iso), + } + if terms.get("endDate"): + args["endDate"] = terms["endDate"] + if overrides: + args.update(overrides) + + raw = await self._call_tool("deals_create", args) + return self._parse_deal(raw) + + async def update_deal(self, deal_id: str, updates: dict[str, Any]) -> SSPDeal: + raw = await self._call_tool("deals_update", {"id": deal_id, **updates}) + return self._parse_deal(raw) + + async def troubleshoot_deal(self, deal_id: str) -> SSPTroubleshootResult: + raw = await self._call_tool("deals_status", {"dealId": deal_id}) + + issues: list[str] = [] + if isinstance(raw, dict): + # deals_status returns labeled summaries under raw["status"]["buyerStatuses"] + # (not raw["buyerSeats"]) — each entry has "status" (label) and "seatId" + status_block = raw.get("status", {}) + for seat in status_block.get("buyerStatuses", []): + if isinstance(seat, dict) and seat.get("status") == "Rejected": + issues.append(f"Buyer seat {seat.get('seatId', '?')} was rejected by provider") + + return SSPTroubleshootResult( + deal_id=deal_id, + status=self._seller_status_label(raw), + primary_issues=issues, + raw=raw, + ) + + # ── Response Parsing ─────────────────────────────────────────────────── + + def _parse_deal(self, raw: Any) -> SSPDeal: + """Parse a deals-api-mcp tool response into SSPDeal.""" + if not isinstance(raw, dict): + return SSPDeal(deal_id="unknown", ssp_name=self.ssp_name) + + # deals_create: {"success": true, "deal": {...}} + # deals_status: {"success": true, "deal": {...}, "status": {...}} + deal = raw.get("deal", raw) + if not isinstance(deal, dict): + deal = raw + + terms = deal.get("terms", {}) if isinstance(deal.get("terms"), dict) else {} + seller_status_int = deal.get("sellerStatus") + + return SSPDeal( + # Internal UUID is what all MCP tools accept (z.string().uuid()). + # external_deal_id is the OpenRTB / IAB ID for DSP activation. + deal_id=str(deal.get("id", "unknown")), + external_deal_id=deal.get("externalDealId"), + name=deal.get("name"), + status=_SELLER_STATUS_MAP.get(seller_status_int, SSPDealStatus.CREATED), + cpm=terms.get("dealFloor"), + currency=terms.get("currency", "USD"), + ssp_name=self.ssp_name, + raw=raw, + ) + + def _seller_status_label(self, raw: Any) -> str: + if isinstance(raw, dict): + # deals_status provides a pre-labeled status block — prefer it + status_block = raw.get("status", {}) + if isinstance(status_block, dict) and status_block.get("sellerStatus"): + return status_block["sellerStatus"] + deal = raw.get("deal", raw) + if isinstance(deal, dict): + code = deal.get("sellerStatus") + return _SELLER_STATUS_LABEL.get(code, "unknown") + return "unknown" diff --git a/src/ad_seller/clients/ssp_base.py b/src/ad_seller/clients/ssp_base.py index 3713545..cffb512 100644 --- a/src/ad_seller/clients/ssp_base.py +++ b/src/ad_seller/clients/ssp_base.py @@ -38,6 +38,7 @@ class SSPType(str, Enum): INDEX_EXCHANGE = "index_exchange" OPENX = "openx" XANDR = "xandr" + DEAL_SYNC = "deal_sync" # IAB Deal Sync API hub (e.g. deals-api-mcp) CUSTOM = "custom" @@ -65,6 +66,7 @@ class SSPDeal(BaseModel): """Normalized deal representation from an SSP.""" deal_id: str + external_deal_id: Optional[str] = None # OpenRTB / IAB deal ID (e.g. "IAB-...") name: Optional[str] = None deal_type: SSPDealType = SSPDealType.PMP status: SSPDealStatus = SSPDealStatus.CREATED diff --git a/src/ad_seller/config/settings.py b/src/ad_seller/config/settings.py index d481bb8..590ff07 100644 --- a/src/ad_seller/config/settings.py +++ b/src/ad_seller/config/settings.py @@ -159,6 +159,14 @@ class Settings(BaseSettings): index_exchange_api_url: Optional[str] = None index_exchange_api_key: Optional[str] = None + # Deal Sync Connectors (external deal-sync services, peer of SSP connectors) + # Comma-separated list of provider names to enable + deal_sync_connectors: str = "" # e.g. "deals_api_mcp" (env: DEAL_SYNC_CONNECTORS) + # IAB Deals MCP (deals-api-mcp server — HTTP Streamable transport) + deals_api_mcp_url: Optional[str] = None # e.g. http://localhost:3100/mcp + deals_api_mcp_key: Optional[str] = None # IAB_DEALS_API_KEY on the MCP server + deals_api_mcp_seller_origin: str = "publisher.example.com" # origin field for deals_create + # Pricing Configuration default_currency: str = "USD" min_deal_value: float = 1000.0 diff --git a/src/ad_seller/flows/execution_activation_flow.py b/src/ad_seller/flows/execution_activation_flow.py index 2327dad..0e9ac44 100644 --- a/src/ad_seller/flows/execution_activation_flow.py +++ b/src/ad_seller/flows/execution_activation_flow.py @@ -259,17 +259,9 @@ async def distribute_to_ssps(self) -> None: settings = get_settings() - if not settings.ssp_connectors: - return # No SSPs configured — skip - from ..clients.ssp_base import SSPDealCreateRequest, SSPDealType - from ..clients.ssp_factory import build_ssp_registry - - registry = build_ssp_registry(settings) - if not registry.list_ssps(): - return - # Map deal type + # Map deal type once — used by both SSP and deal-sync blocks deal_type_str = ( deal.deal_type.value if hasattr(deal.deal_type, "value") else str(deal.deal_type) ) @@ -285,32 +277,74 @@ async def distribute_to_ssps(self) -> None: cpm=deal.price, ) - # Route to appropriate SSP - ssp = registry.get_client_for( - inventory_type=getattr(deal, "product_id", None), - deal_type=deal_type_str, - ) - - async with ssp: - ssp_result = await ssp.create_deal(create_req) - - self.state.execution_orders.setdefault(deal.deal_id, {})["ssp_deal"] = { - "ssp_name": ssp_result.ssp_name, - "ssp_deal_id": ssp_result.deal_id, - "ssp_status": ssp_result.status.value, - } - - await emit_event( - event_type=EventType.DEAL_SYNCED, - flow_id=self.state.flow_id, - flow_type=self.state.flow_type, - deal_id=self.state.deal_id, - payload={"ssp_name": ssp_result.ssp_name, "ssp_deal_id": ssp_result.deal_id}, - ) + # SSP block + if settings.ssp_connectors: + from ..clients.ssp_factory import build_ssp_registry + + ssp_registry = build_ssp_registry(settings) + if ssp_registry.list_ssps(): + ssp = ssp_registry.get_client_for( + inventory_type=getattr(deal, "product_id", None), + deal_type=deal_type_str, + ) + async with ssp: + ssp_result = await ssp.create_deal(create_req) + + self.state.execution_orders.setdefault(deal.deal_id, {})["ssp_deal"] = { + "channel": "ssp", + "ssp_name": ssp_result.ssp_name, + "ssp_deal_id": ssp_result.deal_id, + "external_deal_id": ssp_result.external_deal_id, + "ssp_status": ssp_result.status.value, + } + + await emit_event( + event_type=EventType.DEAL_SYNCED, + flow_id=self.state.flow_id, + flow_type=self.state.flow_type, + deal_id=self.state.deal_id, + payload={ + "channel": "ssp", + "ssp_name": ssp_result.ssp_name, + "ssp_deal_id": ssp_result.deal_id, + "external_deal_id": ssp_result.external_deal_id, + }, + ) + + # Deal-sync block + if settings.deal_sync_connectors: + from ..clients.deal_sync_factory import build_deal_sync_registry + + sync_registry = build_deal_sync_registry(settings) + if sync_registry.list_providers(): + sync_client = sync_registry.get_default() + async with sync_client: + sync_result = await sync_client.create_deal(create_req) + + self.state.execution_orders.setdefault(deal.deal_id, {})["deal_sync"] = { + "provider": sync_client.provider.value, + "deal_sync_deal_id": sync_result.deal_id, + "external_deal_id": sync_result.external_deal_id, + "status": sync_result.status.value, + } + + await emit_event( + event_type=EventType.DEAL_SYNCED, + flow_id=self.state.flow_id, + flow_type=self.state.flow_type, + deal_id=self.state.deal_id, + payload={ + "channel": "deal_sync", + "provider": sync_client.provider.value, + "provider_name": sync_client.provider_name, + "deal_sync_deal_id": sync_result.deal_id, + "external_deal_id": sync_result.external_deal_id, + }, + ) except Exception as e: - # SSP distribution failure is non-fatal — deal still exists in ad server - self.state.warnings.append(f"SSP distribution failed (non-fatal): {e}") + # Distribution failure is non-fatal — deal still exists in ad server + self.state.warnings.append(f"Deal distribution failed (non-fatal): {e}") @listen(distribute_to_ssps) async def update_execution_status(self) -> None: diff --git a/src/ad_seller/services/deal_service.py b/src/ad_seller/services/deal_service.py index d0b38bd..c12311a 100644 --- a/src/ad_seller/services/deal_service.py +++ b/src/ad_seller/services/deal_service.py @@ -968,37 +968,53 @@ async def get_deal_buyer_status(deal_id: str, buyer_url: str) -> dict[str, Any]: async def distribute_deal_via_ssp(request: Any) -> dict[str, Any]: - """Distribute a deal through configured SSP(s).""" + """Distribute a deal through configured SSP(s) or deal-sync providers.""" + from ..clients.deal_sync_factory import build_deal_sync_registry from ..clients.ssp_base import SSPDealCreateRequest, SSPDealType from ..clients.ssp_factory import build_ssp_registry - registry = build_ssp_registry() + ssp_registry = build_ssp_registry() + sync_registry = build_deal_sync_registry() - if not registry.list_ssps(): + if not ssp_registry.list_ssps() and not sync_registry.list_providers(): raise HTTPException( status_code=503, detail={ - "error": "no_ssps_configured", - "message": "No SSP connectors configured. Set SSP_CONNECTORS in environment.", + "error": "no_connectors_configured", + "message": ( + "No connectors configured. " + "Set SSP_CONNECTORS or DEAL_SYNC_CONNECTORS in environment." + ), }, ) - # Get the right SSP client + # Resolve client — try SSP first, then deal-sync + client: Any = None + is_deal_sync = False try: if request.ssp_name: - ssp = registry.get_client(request.ssp_name) + try: + client = ssp_registry.get_client(request.ssp_name) + except KeyError: + client = sync_registry.get_client(request.ssp_name) + is_deal_sync = True else: - ssp = registry.get_client_for( - inventory_type=request.inventory_type, - deal_type=request.deal_type, - ) + if ssp_registry.list_ssps(): + client = ssp_registry.get_client_for( + inventory_type=request.inventory_type, + deal_type=request.deal_type, + ) + else: + client = sync_registry.get_default() + is_deal_sync = True except (KeyError, RuntimeError) as e: raise HTTPException( status_code=400, detail={ "error": "ssp_routing_failed", "message": str(e), - "available_ssps": registry.list_ssps(), + "available_ssps": ssp_registry.list_ssps(), + "available_deal_sync_providers": sync_registry.list_providers(), }, ) @@ -1023,11 +1039,27 @@ async def distribute_deal_via_ssp(request: Any) -> dict[str, Any]: targeting=request.targeting, ) - async with ssp: - result = await ssp.create_deal(create_request) + try: + async with client: + result = await client.create_deal(create_request) + except ValueError as e: + raise HTTPException(status_code=400, detail={"error": "invalid_request", "message": str(e)}) + + if is_deal_sync: + return { + "deal_id": result.deal_id, + "external_deal_id": result.external_deal_id, + "channel": client.channel, + "provider": client.provider.value, + "provider_name": client.provider_name, + "status": result.status.value, + "deal": result.model_dump(exclude={"raw", "ssp_type", "ssp_name"}), + } return { "deal_id": result.deal_id, + "external_deal_id": result.external_deal_id, + "channel": "ssp", "ssp": result.ssp_name, "ssp_type": result.ssp_type.value, "status": result.status.value, @@ -1036,27 +1068,40 @@ async def distribute_deal_via_ssp(request: Any) -> dict[str, Any]: async def troubleshoot_deal_via_ssp(deal_id: str, ssp_name: str) -> dict[str, Any]: - """Troubleshoot a deal via SSP diagnostics.""" + """Troubleshoot a deal via SSP or deal-sync provider diagnostics.""" + from ..clients.deal_sync_factory import build_deal_sync_registry from ..clients.ssp_factory import build_ssp_registry - registry = build_ssp_registry() + ssp_registry = build_ssp_registry() + sync_registry = build_deal_sync_registry() + # Try SSP first, then deal-sync + client: Any = None try: - ssp = registry.get_client(ssp_name) + client = ssp_registry.get_client(ssp_name) except KeyError: - raise HTTPException( - status_code=400, - detail={ - "error": "unknown_ssp", - "message": f"SSP '{ssp_name}' not configured.", - "available_ssps": registry.list_ssps(), - }, - ) + try: + client = sync_registry.get_client(ssp_name) + except KeyError: + raise HTTPException( + status_code=400, + detail={ + "error": "unknown_ssp", + "message": f"SSP '{ssp_name}' not configured.", + "available_ssps": ssp_registry.list_ssps(), + "available_deal_sync_providers": sync_registry.list_providers(), + }, + ) + + async with client: + result = await client.troubleshoot_deal(deal_id) - async with ssp: - result = await ssp.troubleshoot_deal(deal_id) + from ..clients.deal_sync_base import DealSyncClient - return result.model_dump(exclude={"raw"}) + # SSP clients: preserve ssp_type (existing field); deal-sync clients: exclude it + # (SSPDeal.ssp_type defaults to "custom" for deal-sync — not meaningful to callers) + exclude: set[str] = {"raw", "ssp_type"} if isinstance(client, DealSyncClient) else {"raw"} + return result.model_dump(exclude=exclude) # ============================================================================= diff --git a/tests/unit/test_deals_api_mcp_client.py b/tests/unit/test_deals_api_mcp_client.py new file mode 100644 index 0000000..b5d776e --- /dev/null +++ b/tests/unit/test_deals_api_mcp_client.py @@ -0,0 +1,706 @@ +# Author: Green Mountain Systems AI Inc. +# Donated to IAB Tech Lab + +"""Unit tests for DealsAPIMCPClient. + +All tests mock FreeWheelMCPClient.call_tool so no live server is required. +""" + +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from ad_seller.clients.deals_api_mcp_client import ( + _SELLER_STATUS_MAP, + DealsAPIMCPClient, +) +from ad_seller.clients.ssp_base import ( + SSPDeal, + SSPDealCreateRequest, + SSPDealStatus, + SSPDealType, + SSPType, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_client() -> DealsAPIMCPClient: + client = DealsAPIMCPClient( + mcp_url="http://localhost:3100/mcp", + seller_origin="test.example.com", + ) + client._mcp_client = MagicMock() + client._mcp_client.call_tool = AsyncMock() + return client + + +def _deal_response( + *, + deal_id: str = "ext-123", + internal_id: str = "abc", + name: str = "Test Deal", + seller_status: int = 2, + floor: float = 10.0, + currency: str = "USD", +) -> dict: + """Build a minimal deals_create / deals_status response payload.""" + return { + "success": True, + "deal": { + "id": internal_id, + "externalDealId": deal_id, + "name": name, + "sellerStatus": seller_status, + "seller": "Test Publisher", + "terms": { + "dealFloor": floor, + "currency": currency, + "startDate": "2026-01-01T00:00:00Z", + }, + }, + } + + +# --------------------------------------------------------------------------- +# _parse_deal +# --------------------------------------------------------------------------- + + +class TestParseDeal: + def setup_method(self): + self.client = _make_client() + + def test_extracts_internal_uuid_as_deal_id(self): + # deal_id stores internal UUID (used by all MCP tools); external_deal_id is for DSP + raw = _deal_response(internal_id="uuid-abc", deal_id="IAB-999", seller_status=0) + deal = self.client._parse_deal(raw) + assert deal.deal_id == "uuid-abc" + assert deal.external_deal_id == "IAB-999" + + def test_uses_internal_id_field(self): + raw = {"success": True, "deal": {"id": "fallback-id", "sellerStatus": 2, "terms": {}}} + deal = self.client._parse_deal(raw) + assert deal.deal_id == "fallback-id" + assert deal.external_deal_id is None + + def test_maps_seller_status_0_to_active(self): + deal = self.client._parse_deal(_deal_response(seller_status=0)) + assert deal.status == SSPDealStatus.ACTIVE + + def test_maps_seller_status_1_to_paused(self): + deal = self.client._parse_deal(_deal_response(seller_status=1)) + assert deal.status == SSPDealStatus.PAUSED + + def test_maps_seller_status_2_to_created(self): + deal = self.client._parse_deal(_deal_response(seller_status=2)) + assert deal.status == SSPDealStatus.CREATED + + def test_maps_seller_status_4_to_expired(self): + deal = self.client._parse_deal(_deal_response(seller_status=4)) + assert deal.status == SSPDealStatus.EXPIRED + + def test_maps_seller_status_5_to_archived(self): + deal = self.client._parse_deal(_deal_response(seller_status=5)) + assert deal.status == SSPDealStatus.ARCHIVED + + def test_unknown_seller_status_defaults_to_created(self): + deal = self.client._parse_deal(_deal_response(seller_status=99)) + assert deal.status == SSPDealStatus.CREATED + + def test_extracts_floor_and_currency(self): + deal = self.client._parse_deal(_deal_response(floor=25.5, currency="EUR")) + assert deal.cpm == 25.5 + assert deal.currency == "EUR" + + def test_returns_unknown_for_non_dict_input(self): + deal = self.client._parse_deal("not a dict") + assert deal.deal_id == "unknown" + assert deal.ssp_type == SSPType.CUSTOM # SSPDeal model default + + def test_handles_missing_terms_gracefully(self): + raw = {"deal": {"id": "x", "externalDealId": "ext-x", "sellerStatus": 0}} + deal = self.client._parse_deal(raw) + assert deal.deal_id == "x" # internal UUID + assert deal.external_deal_id == "ext-x" + assert deal.cpm is None + + def test_ssp_name_and_type_always_set(self): + deal = self.client._parse_deal(_deal_response()) + assert deal.ssp_name == "IAB Deals MCP" + assert deal.ssp_type == SSPType.CUSTOM # SSPDeal model default; excluded from API responses + + +# --------------------------------------------------------------------------- +# create_deal +# --------------------------------------------------------------------------- + + +class TestCreateDeal: + def setup_method(self): + self.client = _make_client() + + @pytest.mark.asyncio + async def test_maps_request_fields_to_mcp_args(self): + self.client._mcp_client.call_tool.return_value = _deal_response(internal_id="new-uuid-1", deal_id="IAB-new-1") + + request = SSPDealCreateRequest( + name="Q3 Video", + advertiser="Acme Corp", + cpm=30.0, + start_date="2026-07-01T00:00:00Z", + end_date="2026-09-30T00:00:00Z", + impressions_goal=500_000, + buyer_seat_ids=["seat-a", "seat-b"], + currency="GBP", + ) + deal = await self.client.create_deal(request) + + call_args = self.client._mcp_client.call_tool.call_args + tool_name, args = call_args[0] + assert tool_name == "deals_create" + assert args["name"] == "Q3 Video" + assert args["seller"] == "test.example.com" # configured seller origin, not advertiser + assert args["dealFloor"] == 30.0 + assert args["startDate"] == "2026-07-01T00:00:00Z" + assert args["endDate"] == "2026-09-30T00:00:00Z" + assert args["units"] == 500_000 + assert args["wseat"] == ["seat-a", "seat-b"] + assert args["currency"] == "GBP" + assert args["origin"] == "test.example.com" + assert deal.deal_id == "new-uuid-1" # internal UUID + assert deal.external_deal_id == "IAB-new-1" + + @pytest.mark.asyncio + async def test_uses_defaults_for_missing_optional_fields(self): + self.client._mcp_client.call_tool.return_value = _deal_response() + + await self.client.create_deal(SSPDealCreateRequest(cpm=5.0)) + + call_args = self.client._mcp_client.call_tool.call_args + _, args = call_args[0] + assert args["name"] == "Untitled Deal" + assert args["dealFloor"] == 5.0 + assert "endDate" not in args + assert "units" not in args + + @pytest.mark.asyncio + async def test_raises_on_missing_cpm(self): + with pytest.raises(ValueError, match="cpm is required"): + await self.client.create_deal(SSPDealCreateRequest()) + + @pytest.mark.asyncio + async def test_pg_deal_sets_guar_flag(self): + self.client._mcp_client.call_tool.return_value = _deal_response(internal_id="pg-uuid-1") + await self.client.create_deal(SSPDealCreateRequest(cpm=50.0, deal_type=SSPDealType.PG)) + + _, args = self.client._mcp_client.call_tool.call_args[0] + assert args.get("guar") == 1 + + @pytest.mark.asyncio + async def test_non_pg_deal_has_no_guar_flag(self): + self.client._mcp_client.call_tool.return_value = _deal_response(internal_id="pmp-uuid-1") + await self.client.create_deal(SSPDealCreateRequest(cpm=20.0, deal_type=SSPDealType.PMP)) + + _, args = self.client._mcp_client.call_tool.call_args[0] + assert "guar" not in args + + @pytest.mark.asyncio + async def test_returns_ssp_deal_instance(self): + self.client._mcp_client.call_tool.return_value = _deal_response(internal_id="ret-uuid-1") + result = await self.client.create_deal(SSPDealCreateRequest(cpm=10.0)) + assert isinstance(result, SSPDeal) + + +# --------------------------------------------------------------------------- +# get_deal +# --------------------------------------------------------------------------- + + +class TestGetDeal: + def setup_method(self): + self.client = _make_client() + + @pytest.mark.asyncio + async def test_calls_deals_status_with_deal_id(self): + self.client._mcp_client.call_tool.return_value = _deal_response(internal_id="uuid-42", deal_id="IAB-42") + deal = await self.client.get_deal("uuid-42") + + self.client._mcp_client.call_tool.assert_called_once_with( + "deals_status", {"dealId": "uuid-42"} + ) + assert deal.deal_id == "uuid-42" # internal UUID returned by deals_status + assert deal.external_deal_id == "IAB-42" + + +# --------------------------------------------------------------------------- +# list_deals +# --------------------------------------------------------------------------- + + +class TestListDeals: + def setup_method(self): + self.client = _make_client() + + @pytest.mark.asyncio + async def test_returns_list_of_ssp_deals(self): + self.client._mcp_client.call_tool.return_value = { + "deals": [ + { + "id": "i1", + "externalDealId": "e1", + "sellerStatus": 0, + "name": "Deal A", + "terms": {"dealFloor": 5.0, "currency": "USD"}, + }, + { + "id": "i2", + "externalDealId": "e2", + "sellerStatus": 1, + "name": "Deal B", + "terms": {"dealFloor": 8.0, "currency": "USD"}, + }, + ] + } + + deals = await self.client.list_deals() + assert len(deals) == 2 + assert deals[0].deal_id == "i1" # internal UUID + assert deals[0].external_deal_id == "e1" + assert deals[0].status == SSPDealStatus.ACTIVE + assert deals[1].deal_id == "i2" # internal UUID + assert deals[1].external_deal_id == "e2" + assert deals[1].status == SSPDealStatus.PAUSED + + @pytest.mark.asyncio + async def test_caps_page_size_at_100(self): + self.client._mcp_client.call_tool.return_value = {"deals": []} + await self.client.list_deals(limit=999) + + _, args = self.client._mcp_client.call_tool.call_args[0] + assert args["pageSize"] == 100 + + @pytest.mark.asyncio + async def test_passes_status_filter_to_mcp(self): + self.client._mcp_client.call_tool.return_value = {"deals": []} + await self.client.list_deals(status=SSPDealStatus.ACTIVE) + + _, args = self.client._mcp_client.call_tool.call_args[0] + assert args["status"] == 0 # ACTIVE maps to status=0 (deals-api-mcp schema) + + @pytest.mark.asyncio + async def test_no_status_filter_omits_status_param(self): + self.client._mcp_client.call_tool.return_value = {"deals": []} + await self.client.list_deals() + + _, args = self.client._mcp_client.call_tool.call_args[0] + assert "status" not in args + + @pytest.mark.asyncio + async def test_returns_empty_list_on_non_dict_response(self): + self.client._mcp_client.call_tool.return_value = "unexpected" + result = await self.client.list_deals() + assert result == [] + + +# --------------------------------------------------------------------------- +# clone_deal +# --------------------------------------------------------------------------- + + +class TestCloneDeal: + def setup_method(self): + self.client = _make_client() + + @pytest.mark.asyncio + async def test_prefixes_name_with_copy_of(self): + source = _deal_response(internal_id="src-uuid-1", deal_id="IAB-src-1", name="Original Deal", floor=20.0) + new_deal = _deal_response(internal_id="clone-uuid-1", deal_id="IAB-clone-1", name="Copy of Original Deal") + + # clone_deal: (1) deals_status for source, (2) deals_create + self.client._mcp_client.call_tool.side_effect = [source, new_deal] + await self.client.clone_deal("src-uuid-1") + + create_call = self.client._mcp_client.call_tool.call_args_list[1] + _, args = create_call[0] + assert args["name"] == "Copy of Original Deal" + assert args["dealFloor"] == 20.0 + + @pytest.mark.asyncio + async def test_applies_overrides(self): + source = _deal_response(internal_id="src-uuid-2", deal_id="IAB-src-2", floor=10.0) + new_deal = _deal_response(internal_id="clone-uuid-2", deal_id="IAB-clone-2") + + # clone_deal: (1) deals_status for source, (2) deals_create + self.client._mcp_client.call_tool.side_effect = [source, new_deal] + await self.client.clone_deal("src-uuid-2", overrides={"dealFloor": 99.0, "name": "Custom"}) + + create_call = self.client._mcp_client.call_tool.call_args_list[1] + _, args = create_call[0] + assert args["dealFloor"] == 99.0 + assert args["name"] == "Custom" + + @pytest.mark.asyncio + async def test_raises_when_source_has_no_floor(self): + source = { + "success": True, + "deal": { + "id": "src-uuid-3", + "externalDealId": "IAB-src-3", + "name": "No Floor", + "sellerStatus": 2, + "terms": {"startDate": "2026-01-01T00:00:00Z"}, + }, + } + self.client._mcp_client.call_tool.side_effect = [source] + + with pytest.raises(ValueError, match="dealFloor is required"): + await self.client.clone_deal("src-uuid-3") + + # Must not call deals_create after the validation failure + assert self.client._mcp_client.call_tool.call_count == 1 + + @pytest.mark.asyncio + async def test_override_floor_allows_clone_without_source_floor(self): + source = { + "success": True, + "deal": { + "id": "src-uuid-4", + "externalDealId": "IAB-src-4", + "name": "No Floor", + "sellerStatus": 2, + "terms": {"startDate": "2026-01-01T00:00:00Z"}, + }, + } + new_deal = _deal_response(internal_id="clone-uuid-4", deal_id="IAB-clone-4") + self.client._mcp_client.call_tool.side_effect = [source, new_deal] + + await self.client.clone_deal("src-uuid-4", overrides={"dealFloor": 12.0}) + + create_call = self.client._mcp_client.call_tool.call_args_list[1] + _, args = create_call[0] + assert args["dealFloor"] == 12.0 + + +# --------------------------------------------------------------------------- +# update_deal +# --------------------------------------------------------------------------- + + +class TestUpdateDeal: + def setup_method(self): + self.client = _make_client() + + @pytest.mark.asyncio + async def test_passes_id_and_updates_to_mcp(self): + self.client._mcp_client.call_tool.return_value = { + "deal": {"id": "u1", "externalDealId": "ext-u1", "sellerStatus": 0, "terms": {"dealFloor": 12.0}} + } + await self.client.update_deal("u1", {"name": "Updated Name", "dealFloor": 12.0}) + + tool_name, args = self.client._mcp_client.call_tool.call_args[0] + assert tool_name == "deals_update" + assert args["id"] == "u1" + assert args["name"] == "Updated Name" + assert args["dealFloor"] == 12.0 + + @pytest.mark.asyncio + async def test_returns_updated_ssp_deal(self): + self.client._mcp_client.call_tool.return_value = { + "deal": {"id": "u2", "externalDealId": "ext-u2", "sellerStatus": 0, "terms": {"dealFloor": 5.0}} + } + result = await self.client.update_deal("u2", {"dealFloor": 5.0}) + + assert result.deal_id == "u2" + assert result.external_deal_id == "ext-u2" + assert result.cpm == 5.0 + assert result.status == SSPDealStatus.ACTIVE + + +# --------------------------------------------------------------------------- +# troubleshoot_deal +# --------------------------------------------------------------------------- + + +class TestTroubleshootDeal: + def setup_method(self): + self.client = _make_client() + + @pytest.mark.asyncio + async def test_reports_rejected_buyer_seats(self): + # deals_status returns labeled summaries under status.buyerStatuses, not top-level buyerSeats + self.client._mcp_client.call_tool.return_value = { + "deal": {"id": "d1", "externalDealId": "e1", "sellerStatus": 0, "terms": {}}, + "status": { + "sellerStatus": "Active", + "buyerStatuses": [ + {"seatId": "seat-1", "providerId": "mock", "status": "Rejected", "platformDealId": None}, + {"seatId": "seat-2", "providerId": "mock", "status": "Approved", "platformDealId": None}, + ], + }, + } + + result = await self.client.troubleshoot_deal("d1") + assert len(result.primary_issues) == 1 + assert "seat-1" in result.primary_issues[0] + + @pytest.mark.asyncio + async def test_no_issues_when_all_seats_approved(self): + self.client._mcp_client.call_tool.return_value = { + "deal": {"id": "d1", "externalDealId": "e1", "sellerStatus": 0, "terms": {}}, + "status": { + "sellerStatus": "Active", + "buyerStatuses": [ + {"seatId": "seat-1", "providerId": "mock", "status": "Approved", "platformDealId": None}, + ], + }, + } + + result = await self.client.troubleshoot_deal("d1") + assert result.primary_issues == [] + + @pytest.mark.asyncio + async def test_deal_id_preserved_in_result(self): + self.client._mcp_client.call_tool.return_value = { + "deal": {"id": "target-deal", "sellerStatus": 2, "terms": {}}, + "status": {"sellerStatus": "Pending", "buyerStatuses": []}, + } + + result = await self.client.troubleshoot_deal("target-deal") + assert result.deal_id == "target-deal" + + +# --------------------------------------------------------------------------- +# Status map completeness +# --------------------------------------------------------------------------- + + +class TestSellerStatusMap: + def test_all_known_codes_present(self): + # IAB spec codes: 0=Active, 1=Paused, 2=Pending, 4=Complete, 5=Archived + assert set(_SELLER_STATUS_MAP.keys()) == {0, 1, 2, 4, 5} + + def test_code_3_is_not_mapped(self): + # Code 3 is unassigned in the IAB spec — must not silently map to a status + assert 3 not in _SELLER_STATUS_MAP + + +# --------------------------------------------------------------------------- +# DealSyncRegistry factory registration +# --------------------------------------------------------------------------- + + +class TestDealSyncFactory: + def test_registers_deals_api_mcp_when_configured(self): + from unittest.mock import MagicMock + + from ad_seller.clients.deal_sync_factory import build_deal_sync_registry + from ad_seller.clients.deals_api_mcp_client import DealsAPIMCPClient + + settings = MagicMock() + settings.deal_sync_connectors = "deals_api_mcp" + settings.deals_api_mcp_url = "http://localhost:3100/mcp" + settings.deals_api_mcp_key = None + settings.deals_api_mcp_seller_origin = "publisher.example.com" + + registry = build_deal_sync_registry(settings) + assert "deals_api_mcp" in registry.list_providers() + assert isinstance(registry.get_client("deals_api_mcp"), DealsAPIMCPClient) + + def test_registers_nothing_when_url_missing(self): + from unittest.mock import MagicMock + + from ad_seller.clients.deal_sync_factory import build_deal_sync_registry + + settings = MagicMock() + settings.deal_sync_connectors = "deals_api_mcp" + settings.deals_api_mcp_url = None + + registry = build_deal_sync_registry(settings) + assert registry.list_providers() == [] + + def test_empty_connectors_returns_empty_registry(self): + from unittest.mock import MagicMock + + from ad_seller.clients.deal_sync_factory import build_deal_sync_registry + + settings = MagicMock() + settings.deal_sync_connectors = "" + + registry = build_deal_sync_registry(settings) + assert registry.list_providers() == [] + + +# --------------------------------------------------------------------------- +# Persistent MCP session lifecycle +# --------------------------------------------------------------------------- + + +def _reset_session_state() -> None: + """Tear down class-level persistent session state between tests.""" + cls = DealsAPIMCPClient + if cls._session_done is not None: + cls._session_done.set() + task = cls._session_task + if task is not None and not task.done(): + task.cancel() + cls._shared_mcp = None + cls._session_task = None + cls._session_ready = None + cls._session_done = None + cls._session_error = None + cls._session_url = None + cls._session_lock = None + + +class TestPersistentSession: + def setup_method(self): + _reset_session_state() + + def teardown_method(self): + _reset_session_state() + + @pytest.mark.asyncio + async def test_starts_shared_session_on_first_aenter(self): + from contextlib import asynccontextmanager + from unittest.mock import AsyncMock, patch + + fake_session = MagicMock() + fake_session.initialize = AsyncMock() + fake_session.__aenter__ = AsyncMock(return_value=fake_session) + fake_session.__aexit__ = AsyncMock(return_value=None) + + @asynccontextmanager + async def fake_transport(*_args, **_kwargs): + yield (MagicMock(), MagicMock(), None) + + with ( + patch( + "mcp.client.streamable_http.streamablehttp_client", + side_effect=fake_transport, + ) as transport_mock, + patch("mcp.ClientSession", return_value=fake_session), + ): + client = DealsAPIMCPClient(mcp_url="http://localhost:3100/mcp") + async with client: + assert DealsAPIMCPClient._shared_mcp is not None + assert DealsAPIMCPClient._shared_mcp._connected is True + assert DealsAPIMCPClient._session_task is not None + assert not DealsAPIMCPClient._session_task.done() + assert client._mcp_client is DealsAPIMCPClient._shared_mcp + + transport_mock.assert_called_once() + fake_session.initialize.assert_awaited_once() + + @pytest.mark.asyncio + async def test_reuses_session_on_second_aenter(self): + from contextlib import asynccontextmanager + from unittest.mock import AsyncMock, patch + + fake_session = MagicMock() + fake_session.initialize = AsyncMock() + fake_session.__aenter__ = AsyncMock(return_value=fake_session) + fake_session.__aexit__ = AsyncMock(return_value=None) + + @asynccontextmanager + async def fake_transport(*_args, **_kwargs): + yield (MagicMock(), MagicMock(), None) + + with ( + patch( + "mcp.client.streamable_http.streamablehttp_client", + side_effect=fake_transport, + ) as transport_mock, + patch("mcp.ClientSession", return_value=fake_session), + ): + client_a = DealsAPIMCPClient(mcp_url="http://localhost:3100/mcp") + client_b = DealsAPIMCPClient(mcp_url="http://localhost:3100/mcp") + + async with client_a: + first_task = DealsAPIMCPClient._session_task + + async with client_b: + assert DealsAPIMCPClient._session_task is first_task + assert client_b._mcp_client is DealsAPIMCPClient._shared_mcp + + transport_mock.assert_called_once() + fake_session.initialize.assert_awaited_once() + + @pytest.mark.asyncio + async def test_latches_and_reraises_session_start_error(self): + from contextlib import asynccontextmanager + from unittest.mock import patch + + boom = RuntimeError("initialize failed") + + @asynccontextmanager + async def failing_transport(*_args, **_kwargs): + raise boom + yield # pragma: no cover + + with patch( + "mcp.client.streamable_http.streamablehttp_client", + side_effect=failing_transport, + ): + client = DealsAPIMCPClient(mcp_url="http://localhost:3100/mcp") + with pytest.raises(RuntimeError, match="initialize failed"): + async with client: + pass + + assert DealsAPIMCPClient._session_error is boom + + # A later enter retries start; with the same transport failure it + # latches and raises again. + with pytest.raises(RuntimeError, match="initialize failed"): + async with DealsAPIMCPClient(mcp_url="http://localhost:3100/mcp"): + pass + assert DealsAPIMCPClient._session_error is boom + + @pytest.mark.asyncio + async def test_reconnects_after_session_drop(self): + from contextlib import asynccontextmanager + from unittest.mock import AsyncMock, patch + + fake_session = MagicMock() + fake_session.initialize = AsyncMock() + fake_session.__aenter__ = AsyncMock(return_value=fake_session) + fake_session.__aexit__ = AsyncMock(return_value=None) + + @asynccontextmanager + async def fake_transport(*_args, **_kwargs): + yield (MagicMock(), MagicMock(), None) + + with ( + patch( + "mcp.client.streamable_http.streamablehttp_client", + side_effect=fake_transport, + ) as transport_mock, + patch("mcp.ClientSession", return_value=fake_session), + ): + client = DealsAPIMCPClient(mcp_url="http://localhost:3100/mcp") + async with client: + first_task = DealsAPIMCPClient._session_task + first_mcp = DealsAPIMCPClient._shared_mcp + assert first_task is not None + assert first_mcp is not None + assert first_mcp._connected is True + + # Simulate a dropped connection: background task exits. + assert DealsAPIMCPClient._session_done is not None + DealsAPIMCPClient._session_done.set() + await first_task + assert first_task.done() + assert first_mcp._connected is False + + async with DealsAPIMCPClient(mcp_url="http://localhost:3100/mcp") as client_b: + assert DealsAPIMCPClient._session_task is not first_task + assert not DealsAPIMCPClient._session_task.done() + assert DealsAPIMCPClient._shared_mcp is not first_mcp + assert DealsAPIMCPClient._shared_mcp._connected is True + assert client_b._mcp_client is DealsAPIMCPClient._shared_mcp + assert DealsAPIMCPClient._session_error is None + + assert transport_mock.call_count == 2 + assert fake_session.initialize.await_count == 2