voc_full

Instructions

Implementation Reference

• mcp_server/tools.py:210-220 (handler)

  The core handler for voc_full: validates ASIN, runs voc.sh script via _run_script, and parses the markdown output via _parse_analyze_markdown. Returns a dict with asin, market, report_markdown, sentiment, pain_points, selling_points, tips, summary_zh, summary_en.

  def voc_full(asin: str, market: str = "US", limit: int = 100) -> dict[str, Any]:
      """One-shot: scrape + analyze. Equivalent to `bash voc.sh ASIN`."""
      asin = _validate_asin(asin)
      if limit < 1 or limit > 1000:
          raise ValueError(f"limit must be 1-1000, got {limit}")
  
      report_md = _run_script(
          "voc.sh",
          [asin, "--limit", str(limit), "--market", market],
      )
      return _parse_analyze_markdown(asin, market, report_md)

• mcp_server/server.py:72-87 (registration)

  MCP tool registration via @mcp.tool() decorator. The voc_full function is declared as an MCP tool that delegates to tools.voc_full().

  @mcp.tool()
  def voc_full(asin: str, market: str = "US", limit: int = 100) -> dict:
      """One-shot: fetch reviews AND run AI analysis.
  
      The default tool for "give me a VOC report on this ASIN" style requests.
      Internally equivalent to `bash voc.sh ASIN` — calls fetch.sh and
      analyze.sh in sequence.
  
      Args:
          asin: 10-character ASIN.
          market: Market code or amazon.* domain (default: US).
          limit: Number of reviews to fetch (default 100, max 1000).
  
      Returns: Same shape as `analyze_reviews`.
      """
      return tools.voc_full(asin=asin, market=market, limit=limit)

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

  The AnalyzeReport Pydantic schema defines the return type structure for voc_full (and analyze_reviews), with fields: asin, market, report_markdown, sentiment, pain_points, selling_points, tips, summary_zh, summary_en.

  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 = ""

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

  Helper function _validate_asin used by voc_full 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/tools.py:124-171 (helper)

  Helper function _parse_analyze_markdown used by voc_full to extract structured fields from the analyze.sh markdown output.

  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", ""),
      }