client.events: track your users' events, and list them by stream

.claude/skills/portal-app/SKILL.md

@@ -45,6 +45,8 @@ for item in res.items:
     print(item.catalog_id, item.data["name"])
 ```
 
+For an app that reads events, read each stream with `client.events.list(stream=...)`, or every stream when `streams` is missing; `user_id=` narrows it to one shopper. `client.complete(user_id=...)` ranks for a shopper from their stored events.
+
 With one id, `catalog_ids=[cat]` and `catalog_id=cat` are the same. If the app opens on no catalog and none arrives, fall back to `UNBOXAI_CATALOG_ID` or ask; don't guess an id. The API refuses a catalog the key can't read and catalogs that can't be ranked together; show its message rather than an empty result.
 
 ## Workflow

README.md

@@ -102,6 +102,18 @@ res = client.complete([View("paB08NYK61PJ")], user=shopper)
 
 If your site already sends events (the portal's browser script, or `POST /events`), pass the person's id instead of a history: `client.complete(user_id=uid, user=shopper)`. The API adds the events it has stored for that id to whatever `history` you pass. With the browser script, the id is in the `unbox_uid` cookie.
 
+From your server, record events with `client.events.track`, and read back what your account has stored with `client.events.list`, newest first:
+
+```python
+client.events.track(uid, View(shoe), user=shopper)
+client.events.track(uid, Order([shoe]), timestamp=paid_at)  # sent later than it happened
+
+client.events.list(user_id=uid, limit=20)
+client.events.list(stream="api", types=["order"])  # what your server sent; a stream is also a publishable key or an import
+```
+
+An event's catalog is `catalog_id=`, else the one an `Item` of a response names, else the client's default. Pass `timestamp=` whenever you send an event after it happened, so the history stays in order.
+
 ## Your own catalog
 
 The model can only return items it knows about. Out of the box the client uses `retail_catalog`, a sample retail catalog. To use the model on your own data, upload your catalog as a parquet file. A catalog doesn't have to be products for sale: artworks, articles, listings or anything else people browse and pick from work the same way.

src/behaviorgpt/_client.py

@@ -14,6 +14,7 @@
 
 from behaviorgpt._exceptions import AuthenticationError, UnboxAIError
 from behaviorgpt.resources.catalogs import AsyncCatalogs, Catalogs
+from behaviorgpt.resources.events import AsyncEvents, Events
 from behaviorgpt.resources.recommendations import AsyncRecommender, Recommender
 from behaviorgpt.resources.requests import UnboxAIRequester
 from behaviorgpt.resources.search import AsyncSearcher, Searcher
@@ -376,6 +377,7 @@ def __init__(
         self.recommendations = Recommender(self._http_client)
         self.search = Searcher(self._http_client)
         self.catalogs = Catalogs(self._http_client)
+        self.events = Events(self._http_client, self)
         self._sections = Sectioner(self._http_client)
 
     def complete(
@@ -752,6 +754,7 @@ def __init__(
         self.recommendations = AsyncRecommender(self._http_client)
         self.search = AsyncSearcher(self._http_client)
         self.catalogs = AsyncCatalogs(self._http_client)
+        self.events = AsyncEvents(self._http_client, self)
         self._sections = AsyncSectioner(self._http_client)
 
     async def __aenter__(self) -> "AsyncUnboxAIClient":

src/behaviorgpt/resources/events.py

@@ -0,0 +1,189 @@
+from datetime import UTC, datetime
+from typing import Optional, Protocol, Sequence
+from uuid import uuid4
+
+import httpx
+
+from behaviorgpt.resources._shared import handle_response
+from behaviorgpt.types import Domains, Order, UserEvent
+from behaviorgpt.types.http import TrackedEvent
+from behaviorgpt.types.user import User
+
+PATH = "/events"
+OWN_PATH = "/events/own"
+# the server's limit
+MAX_LISTED = 200
+
+
+class EventContext(Protocol):
+    domains: Domains
+    timezone: str
+    default_catalog_id: str
+
+
+def _event_catalog(event: UserEvent) -> Optional[str]:
+    """The catalog the event's items say they are in, if they say."""
+    if isinstance(event, Order):
+        named = {entry.catalog_id for entry in event.items or ()} - {None}
+        if len(named) > 1:
+            raise ValueError(
+                "an Order's items are in one catalog; track one Order per catalog"
+            )
+        return next(iter(named), None)
+    return getattr(event, "catalog_id", None)
+
+
+def event_body(
+    context: EventContext,
+    user_id: str,
+    event: UserEvent,
+    *,
+    catalog_id: Optional[str],
+    timestamp: Optional[datetime],
+    user: Optional[User],
+) -> dict:
+    """The body `track` posts for `event`."""
+    if timestamp is None:
+        timestamp = datetime.now(tz=UTC)
+    elif timestamp.tzinfo is None:
+        raise ValueError("timestamp must be timezone-aware")
+
+    payload = event.model_dump(mode="json", exclude_none=True)
+    if isinstance(event, Order):
+        if not event.items:
+            raise ValueError("track an Order with the items bought: Order([item, ...])")
+        # The API stores orders with a cart id and a unit price per item; the
+        # model reads only which items were bought.
+        payload["cart_id"] = event.cart_id or str(uuid4())
+        payload["items"] = [
+            {
+                "product": entry.item,
+                "quantity": entry.quantity,
+                "unit_selling_price": 0.0,
+            }
+            for entry in event.items
+        ]
+
+    domains = user.domains(context.domains.market) if user else context.domains
+    return {
+        "user_id": user_id,
+        "store_id": catalog_id or _event_catalog(event) or context.default_catalog_id,
+        "event": payload,
+        "domains": domains.model_dump(mode="json", exclude_none=True),
+        "timezone": (user.timezone if user else None) or context.timezone,
+        "timestamp": timestamp.isoformat(),
+    }
+
+
+def list_params(
+    user_id: Optional[str],
+    limit: int,
+    types: Optional[Sequence[str]],
+    stream: Optional[str],
+) -> list[tuple[str, str]]:
+    if not 1 <= limit <= MAX_LISTED:
+        raise ValueError(f"limit must be between 1 and {MAX_LISTED}, got {limit}")
+    params: list[tuple[str, str]] = [("limit", str(limit))]
+    if user_id is not None:
+        params.append(("user_id", user_id))
+    params.extend(("type", t) for t in types or ())
+    if stream is not None:
+        params.append(("stream", stream))
+    return params
+
+
+def tracked_events(data: dict) -> list[TrackedEvent]:
+    return [TrackedEvent(**entry) for entry in data["events"]]
+
+
+class Events:
+    """Your users' events. Record them with `track`; `complete(user_id=...)`
+    and `sections(user_id=...)` then read them as that user's history."""
+
+    def __init__(self, http_client: httpx.Client, context: EventContext):
+        self._client = http_client
+        self._context = context
+
+    def track(
+        self,
+        user_id: str,
+        event: UserEvent,
+        *,
+        catalog_id: Optional[str] = None,
+        timestamp: Optional[datetime] = None,
+        user: Optional[User] = None,
+    ) -> str:
+        """Record that `user_id` did `event`, and return the event's id.
+
+        `timestamp` is when it happened, now by default. Pass it when you send
+        events later than they happen, so that they stay in order. The catalog
+        is `catalog_id`, else the one an `Item` of a response names, else the
+        client's default. `user` says who they are, as in `complete`.
+        """
+        body = event_body(
+            self._context,
+            user_id,
+            event,
+            catalog_id=catalog_id,
+            timestamp=timestamp,
+            user=user,
+        )
+        return handle_response(self._client.post(PATH, json=body))["id"]
+
+    def list(
+        self,
+        *,
+        user_id: Optional[str] = None,
+        limit: int = 50,
+        types: Optional[Sequence[str]] = None,
+        stream: Optional[str] = None,
+    ) -> list[TrackedEvent]:
+        """Your account's latest events, newest first: all your users', or
+        `user_id`'s. `types` keeps only those event types, such as
+        `["viewItem", "search"]`, and `stream` only the events of one stream:
+        a publishable key, "api" for those sent with your key, or an import
+        such as "import_shop". `limit` can be at most 200."""
+        params = list_params(user_id, limit, types, stream)
+        response = self._client.get(OWN_PATH, params=params)
+        return tracked_events(handle_response(response))
+
+
+class AsyncEvents:
+    """`Events` for `AsyncUnboxAIClient`."""
+
+    def __init__(self, http_client: httpx.AsyncClient, context: EventContext):
+        self._client = http_client
+        self._context = context
+
+    async def track(
+        self,
+        user_id: str,
+        event: UserEvent,
+        *,
+        catalog_id: Optional[str] = None,
+        timestamp: Optional[datetime] = None,
+        user: Optional[User] = None,
+    ) -> str:
+        """See `Events.track`."""
+        body = event_body(
+            self._context,
+            user_id,
+            event,
+            catalog_id=catalog_id,
+            timestamp=timestamp,
+            user=user,
+        )
+        return handle_response(await self._client.post(PATH, json=body))["id"]
+
+    async def list(
+        self,
+        *,
+        user_id: Optional[str] = None,
+        limit: int = 50,
+        types: Optional[Sequence[str]] = None,
+        stream: Optional[str] = None,
+    ) -> list[TrackedEvent]:
+        """See `Events.list`."""
+        params = list_params(user_id, limit, types, stream)
+        response = await self._client.get(OWN_PATH, params=params)
+        return tracked_events(handle_response(response))

src/behaviorgpt/types/__init__.py

@@ -28,6 +28,7 @@
     Section,
     SectionsResponse,
     SimilarItemsRequest,
+    TrackedEvent,
     UnboxAIRequest,
     UnboxAIResponse,
 )
@@ -62,6 +63,7 @@
     "CartItem",
     "Section",
     "SectionsResponse",
+    "TrackedEvent",
 ]
 
 

src/behaviorgpt/types/http.py

@@ -90,6 +90,17 @@ def validate_pagination(self) -> "UnboxAIRequest":
         return self
 
 
+class TrackedEvent(BaseModel):
+    """An event as the API stored it: `event` is as the API stores it, and
+    `stream` is where it came from (a publishable key, "api", an import)."""
+
+    user_id: str
+    timestamp: datetime
+    event: dict
+    store_id: Optional[str] = None
+    stream: Optional[str] = None
+
+
 class UnboxAIResponse(BaseModel):
     items: List[Item]
     offset: int