Google Play Console

Instructions

Implementation Reference

• src/google_play_mcp/server.py:801-880 (handler)

  The `get_vitals_summary` function is decorated with `@mcp.tool()` and implements the logic to aggregate crash and ANR rates by version code, check against thresholds, and return a JSON summary.

  @mcp.tool()
  def get_vitals_summary(
      package_name: str,
      days: int = 7,
  ) -> str:
      """Get combined Android Vitals: crash rate and ANR rate per version code.
  
      Returns averages over the period with threshold flags.
      Thresholds: userPerceivedCrashRate > 1.09%, userPerceivedAnrRate > 0.47%.
  
      Args:
          package_name: Package name, e.g. com.example.myapp
          days: Past days to include (default 7, max 30).
      """
      days = max(1, min(days, 30))
      try:
          crash_raw = _reporting().query_crash_rate(package_name, days)
          anr_raw = _reporting().query_anr_rate(package_name, days)
  
          crash_rows = _parse_reporting_rows(crash_raw.get("rows", []))
          anr_rows = _parse_reporting_rows(anr_raw.get("rows", []))
  
          # Aggregate by version code: average rates over the period
          def _aggregate(rows: list, rate_key: str, perceived_key: str) -> dict:
              by_version: dict = {}
              for row in rows:
                  vc = row.get("versionCode") or "unknown"
                  entry = by_version.setdefault(vc, {"values": [], "perceived": [], "users": []})
                  if isinstance(row.get(rate_key), (int, float)):
                      entry["values"].append(row[rate_key])
                  if isinstance(row.get(perceived_key), (int, float)):
                      entry["perceived"].append(row[perceived_key])
                  if isinstance(row.get("distinctUsers"), (int, float)):
                      entry["users"].append(row["distinctUsers"])
              result = {}
              for vc, data in by_version.items():
                  avg = lambda lst: round(sum(lst) / len(lst), 6) if lst else None
                  result[vc] = {
                      f"avg_{rate_key}": avg(data["values"]),
                      f"avg_{perceived_key}": avg(data["perceived"]),
                      "avgDistinctUsers": avg(data["users"]),
                  }
              return result
  
          crash_by_vc = _aggregate(crash_rows, "crashRate", "userPerceivedCrashRate")
          anr_by_vc = _aggregate(anr_rows, "anrRate", "userPerceivedAnrRate")
  
          all_vcs = sorted(
              set(crash_by_vc) | set(anr_by_vc),
              key=lambda x: int(x) if str(x).isdigit() else 0,
              reverse=True,
          )
  
          summary = []
          for vc in all_vcs:
              entry = {"versionCode": vc}
              entry.update(crash_by_vc.get(vc, {}))
              entry.update(anr_by_vc.get(vc, {}))
              # Flag if exceeding bad behavior thresholds
              crash_pct = entry.get("avg_userPerceivedCrashRate")
              anr_pct = entry.get("avg_userPerceivedAnrRate")
              entry["exceedsCrashThreshold"] = crash_pct is not None and crash_pct > 0.0109
              entry["exceedsAnrThreshold"] = anr_pct is not None and anr_pct > 0.0047
              summary.append(entry)
  
          latest = summary[0] if summary else None
  
          return json.dumps(
              {
                  "packageName": package_name,
                  "periodDays": days,
                  "badBehaviorThresholds": {
                      "userPerceivedCrashRate": 0.0109,
                      "userPerceivedAnrRate": 0.0047,
                  },
                  "latestVersionSummary": latest,
                  "allVersions": summary,
              },
              indent=2,
          )