hotmart_subscriber_360_app

Instructions

Implementation Reference

• src/hotmart_mcp/apps/subscriptions.py:188-266 (handler)

  Main handler function for hotmart_subscriber_360_app. Calls Hotmart API to fetch subscriber purchases, computes LTV/total/qty/average ticket, and builds a PrefabApp UI dashboard with Cards (LTV, purchases, average ticket) and a DataTable of purchase history.

  async def hotmart_subscriber_360_app(subscriber_code: str) -> PrefabApp:
      """Visão 360 de um assinante — perfil + histórico de compras.
  
      Cards de LTV/qty/ticket + DataTable com purchases. Use pra 'ver
      dados do assinante X', 'histórico completo do código Y'.
  
      Args:
          subscriber_code: Subscriber code (Hotmart alphanumeric, ex: 'VRWIQQRG').
      """
      client = get_client()
      raw = await client.get(
          f"/payments/api/v1/subscriptions/{subscriber_code}/purchases"
      )
      purchases: list[SubscriberPurchase] = parse_items(raw, SubscriberPurchase)
  
      def _value(p: SubscriberPurchase) -> float:
          if p.price and p.price.value is not None:
              return p.price.value
          return 0.0
  
      total_spent = sum(_value(p) for p in purchases)
      qty = len(purchases)
      sanity_warning = qty > 0 and total_spent == 0
  
      rows = []
      for p in purchases[:100]:
          rows.append({
              "transaction": p.transaction or "?",
              "date": epoch_ms_to_date(p.approved_date),
              "recurrence": str(p.recurrency_number) if p.recurrency_number else "—",
              "value": format_brl(_value(p)),
              "status": (p.status if p.status else None) or "?",
              "payment": (p.payment_type if p.payment_type else None) or "—",
          })
  
      with PrefabApp() as app:
          with Column(gap=4, css_class="p-6"):
              Heading(content=f"Assinante {subscriber_code}")
  
              if sanity_warning:
                  Heading(content="⚠️ LTV zerado com compras no histórico — possível schema drift", level=3)
  
              with Grid(columns=[1, 1, 1], gap=4):
                  with Card():
                      with CardHeader():
                          CardTitle(content="Total gasto (LTV)")
                      with CardContent():
                          Metric(label="LTV", value=format_brl(total_spent))
                  with Card():
                      with CardHeader():
                          CardTitle(content="Total de compras")
                      with CardContent():
                          Metric(label="Compras", value=str(qty))
                  with Card():
                      with CardHeader():
                          CardTitle(content="Ticket médio")
                      with CardContent():
                          Metric(
                              label="Médio",
                              value=format_brl(total_spent / qty) if qty else "—",
                          )
  
              with Card():
                  with CardHeader():
                      CardTitle(content="Histórico de compras")
                  with CardContent():
                      DataTable(
                          columns=[
                              DataTableColumn(key="transaction", header="Transação"),
                              DataTableColumn(key="date", header="Data", sortable=True),
                              DataTableColumn(key="recurrence", header="Recorrência"),
                              DataTableColumn(key="value", header="Valor"),
                              DataTableColumn(key="status", header="Status", sortable=True),
                              DataTableColumn(key="payment", header="Pagamento"),
                          ],
                          rows=rows,
                          search=True,
                      )
      return app

• src/hotmart_mcp/apps/subscriptions.py:269-273 (registration)

  Module-level __all__ export that includes 'hotmart_subscriber_360_app' for auto-discovery by the server.

  __all__ = [
      "hotmart_subscriptions_health_app",
      "hotmart_churn_analyzer_app",
      "hotmart_subscriber_360_app",
  ]

• src/hotmart_mcp/server.py:40-52 (registration)

  Server registration: _discover_and_register_apps() iterates modules under hotmart_mcp.apps and registers any async function ending in '_app' as an MCP tool with app=True.

  def _discover_and_register_apps() -> int:
      """Import all modules under hotmart_mcp.apps and register apps (app=True)."""
      registered = 0
      for module_info in pkgutil.iter_modules(apps_pkg.__path__, prefix=f"{apps_pkg.__name__}."):
          if module_info.name.endswith("__init__"):
              continue
          module = importlib.import_module(module_info.name)
          for name, obj in inspect.getmembers(module, iscoroutinefunction):
              if name.startswith("_") or not name.endswith("_app"):
                  continue
              mcp.tool(app=True)(obj)
              registered += 1
      return registered

• src/hotmart_mcp/apps/subscriptions.py:1-18 (schema)

  Imports for the handler: uses SubscriberPurchase model for type validation and _helpers (format_brl, epoch_ms_to_date, parse_items) for data formatting.

  """Subscriptions dashboards — Prefab UI apps (schema-validated)."""
  from __future__ import annotations
  
  from collections import Counter, defaultdict
  from typing import Optional
  
  from prefab_ui import PrefabApp
  from prefab_ui.components import (
      Card, CardContent, CardHeader, CardTitle, Column, DataTable,
      DataTableColumn, Grid, Heading, Metric,
  )
  from prefab_ui.components.charts import ChartSeries, LineChart, PieChart
  
  from hotmart_mcp._shared import get_client
  from hotmart_mcp.models import Subscription, SubscriberPurchase
  from hotmart_mcp.apps._helpers import (
      epoch_ms_to_date, format_brl, parse_items,
  )

• src/hotmart_mcp/apps/_helpers.py:1-52 (helper)

  Helper utilities used by the 360 app: format_brl (currency formatting), epoch_ms_to_date (timestamp conversion), parse_items (API response normalization), and safe_sum.

  """Shared helpers for Prefab UI apps — schema-validated."""
  from __future__ import annotations
  
  from datetime import datetime
  from typing import Any, TypeVar
  
  from pydantic import BaseModel, TypeAdapter
  
  T = TypeVar("T", bound=BaseModel)
  
  
  def format_brl(value: float | int | None) -> str:
      v = value or 0
      return f"R$ {v:,.2f}".replace(",", "X").replace(".", ",").replace("X", ".")
  
  
  def epoch_ms_to_date(ms: int | None) -> str:
      if not ms:
          return ""
      return datetime.fromtimestamp(ms / 1000).strftime("%Y-%m-%d")
  
  
  def parse_items(raw: Any, model: type[T]) -> list[T]:
      """Validate an API response into a list of typed models.
  
      Hotmart's wrappers vary by endpoint — sometimes `{"items": [...]}`,
      sometimes a raw `[...]`. This normalises both into `list[model]`.
  
      Pydantic raises ValidationError if the response shape no longer
      matches the spec — surfaces schema drift loud instead of silently
      returning empty/zeroed data.
      """
      if isinstance(raw, dict):
          raw_items = raw.get("items", [])
      elif isinstance(raw, list):
          raw_items = raw
      else:
          raw_items = []
      return TypeAdapter(list[model]).validate_python(raw_items)
  
  
  def safe_sum(items, getter, *, label: str = "value") -> float:
      """Sum a numeric attribute from typed items, asserting non-zero
      when count > 0. Returns 0 for empty input. If items exist but
      every value is None/0, returns 0 — caller can render a warning."""
      total = 0.0
      for it in items:
          v = getter(it)
          if v is not None:
              total += v
      return total