analyze_reviews

Instructions

Implementation Reference

• mcp_server/tools.py:176-205 (handler)

  Actual implementation of the analyze_reviews tool. Writes reviews to a temp JSON file, runs analyze.sh shell script, parses the markdown output into structured dict.

  def analyze_reviews(reviews_json: dict[str, Any] | list[dict], asin: str) -> dict[str, Any]:
      """Run AI analysis on a reviews JSON object (or list) that was previously
      fetched. Useful when the caller already has reviews from `fetch_reviews`
      and wants to re-analyze without paying the Shulex API a second time.
      """
      asin = _validate_asin(asin)
  
      # Accept both fetch.sh's `{reviews, meta}` shape and a bare list.
      if isinstance(reviews_json, list):
          wrapped = {"reviews": reviews_json, "meta": {"asin": asin, "market": "US"}}
      else:
          wrapped = reviews_json
          wrapped.setdefault("meta", {}).setdefault("asin", asin)
  
      market = wrapped.get("meta", {}).get("market", "US")
  
      with tempfile.NamedTemporaryFile(
          prefix="mcp_reviews_", suffix=".json", delete=False, mode="w", encoding="utf-8"
      ) as tmp:
          json.dump(wrapped, tmp, ensure_ascii=False)
          reviews_path = tmp.name
      try:
          report_md = _run_script("analyze.sh", [reviews_path, asin])
      finally:
          try:
              os.unlink(reviews_path)
          except OSError:
              pass
  
      return _parse_analyze_markdown(asin, market, report_md)

• mcp_server/server.py:51-69 (registration)

  MCP tool registration via @mcp.tool() decorator. Thin wrapper that delegates to tools.analyze_reviews().

  @mcp.tool()
  def analyze_reviews(reviews_json: dict | list, asin: str) -> dict:
      """Run AI analysis on reviews you already have.
  
      Useful when you fetched reviews via `fetch_reviews` (or your own scraper)
      and want the VOC report — sentiment breakdown, pain points, selling
      points, listing tips — without re-paying the Shulex API.
  
      Args:
          reviews_json: Either fetch.sh's `{reviews, meta}` envelope, or a
              bare list of review objects.
          asin: 10-character ASIN that the reviews belong to (for the report
              header).
  
      Returns:
          {asin, market, report_markdown, sentiment, pain_points,
           selling_points, tips, summary_zh, summary_en}
      """
      return tools.analyze_reviews(reviews_json=reviews_json, asin=asin)

• mcp_server/tools.py:124-171 (helper)

  Parses the markdown output from analyze.sh into structured fields: sentiment counts, pain points, selling points, tips, and summaries in both ZH and EN.

  def _parse_analyze_markdown(asin: str, market: str, report_md: str) -> dict[str, Any]:
      """Pull the structured fields analyze.sh embedded as `KEY: value` lines.
  
      The renderer in analyze.sh emits markers like `SENTIMENT_POSITIVE: 37`,
      `PAIN_POINT_1_ZH: ...`, `TIP_3_EN: ...` interleaved with the prose
      markdown. We grep them out into a flat dict, then assemble structured
      fields. Anything we can't parse stays accessible via `report_markdown`.
      """
      flat: dict[str, str] = {}
      for line in report_md.splitlines():
          m = _LINE_RE.match(line.strip())
          if m:
              flat.setdefault(m.group(1), m.group(2).strip())
  
      def grouped(prefix: str, suffixes: tuple[str, ...]) -> list[dict[str, str]]:
          items = []
          for i in range(1, 11):  # support up to 10; reports typically have 5
              row = {}
              for suf in suffixes:
                  key = f"{prefix}_{i}_{suf}"
                  if key in flat:
                      row[suf.lower()] = flat[key]
              if row:
                  items.append(row)
          return items
  
      sentiment: Optional[dict[str, int]] = None
      if {"SENTIMENT_POSITIVE", "SENTIMENT_NEUTRAL", "SENTIMENT_NEGATIVE"} <= flat.keys():
          try:
              sentiment = {
                  "positive": int(flat["SENTIMENT_POSITIVE"]),
                  "neutral": int(flat["SENTIMENT_NEUTRAL"]),
                  "negative": int(flat["SENTIMENT_NEGATIVE"]),
              }
          except ValueError:
              sentiment = None
  
      return {
          "asin": asin,
          "market": market,
          "report_markdown": report_md,
          "sentiment": sentiment,
          "pain_points": grouped("PAIN_POINT", ("ZH", "EN", "COUNT")),
          "selling_points": grouped("SELLING_POINT", ("ZH", "EN", "COUNT")),
          "tips": grouped("TIP", ("ZH", "EN")),
          "summary_zh": flat.get("SUMMARY_ZH", ""),
          "summary_en": flat.get("SUMMARY_EN", ""),
      }

• mcp_server/tools.py:37-47 (helper)

  Helper used by analyze_reviews to validate and normalize the ASIN parameter.

  def _validate_asin(asin: str) -> str:
      """Validate ASIN shape. We accept lowercase from MCP clients but normalize
      to upper before passing to the shell scripts (which warn but proceed on
      lowercase) — this keeps our error messages clearer.
      """
      asin = asin.strip().upper()
      if not VALID_ASIN_RE.match(asin):
          raise ValueError(
              f"invalid ASIN {asin!r}: must be 10 alphanumeric characters (e.g. B08N5WRWNW)"
          )
      return asin

• mcp_server/schemas.py:38-53 (schema)

  Pydantic model defining the return schema for analyze_reviews output.

  class AnalyzeReport(BaseModel):
      """Report returned by `analyze_reviews` / `voc_full`.
  
      Always includes the raw markdown (`report_markdown`). When the report can
      be parsed cleanly, structured fields are populated; on parse failure the
      structured fields are None and the markdown is still returned verbatim.
      """
      asin: str
      market: str = "US"
      report_markdown: str
      sentiment: Optional[dict] = None
      pain_points: list[dict] = Field(default_factory=list)
      selling_points: list[dict] = Field(default_factory=list)
      tips: list[dict] = Field(default_factory=list)
      summary_zh: str = ""
      summary_en: str = ""