fetch_comments

Instructions

Implementation Reference

• server.py:50-50 (registration)

  The @mcp.tool() decorator registers fetch_comments as an MCP tool on the FastMCP server instance.

  @mcp.tool()

• server.py:51-107 (handler)

  Main handler function that extracts video ID from URL, paginates through YouTube commentThreads API, collects top-level comments and replies, and returns a JSON string with results.

  async def fetch_comments(videoUrl: str, order: str = "relevance", max: int = 300) -> str:
      """
      Fetch public comments for a YouTube video and return a JSON string.
      Args:
        videoUrl: Full YouTube video URL.
        order: "relevance" (default) or "time".
        max: Max total comments to return (100–1000 建議).
      """
      if not os.environ.get("YOUTUBE_API_KEY"):
          return "ERROR: Missing YOUTUBE_API_KEY in environment."
      video_id = _extract_video_id(videoUrl)
      if not video_id:
          return "ERROR: Cannot parse video ID from URL."
      order = order if order in ("relevance", "time") else "relevance"
  
      items: List[Dict[str, Any]] = []
      page_token = None
      try:
          async with httpx.AsyncClient() as client:
              while True:
                  params = {
                      "part": "snippet,replies",
                      "videoId": video_id,
                      "maxResults": 100,
                      "order": order,
                      "textFormat": "plainText",
                  }
                  if page_token:
                      params["pageToken"] = page_token
                  data = await _yt_get(client, "commentThreads", params)
  
                  for th in data.get("items", []):
                      top = th.get("snippet", {}).get("topLevelComment", {})
                      if top:
                          items.append(_pack_comment(top))
                      for rep in th.get("replies", {}).get("comments", []) or []:
                          items.append(_pack_comment(rep, parent_id=top.get("id") if top else None))
  
                      if len(items) >= max:
                          break
                  if len(items) >= max:
                      break
                  page_token = data.get("nextPageToken")
                  if not page_token:
                      break
  
          return json.dumps({
              "video_id": video_id,
              "order": order,
              "requested": max,
              "total_returned": len(items),
              "items": items
          }, ensure_ascii=False)
      except httpx.HTTPStatusError as e:
          return f"ERROR: HTTP {e.response.status_code} – {e.response.text}"
      except Exception as e:
          return f"ERROR: {type(e).__name__}: {e}"

• server.py:51-58 (schema)

  Input schema defined via function signature (videoUrl: str, order: str = "relevance", max: int = 300) and docstring in the tool definition.

  async def fetch_comments(videoUrl: str, order: str = "relevance", max: int = 300) -> str:
      """
      Fetch public comments for a YouTube video and return a JSON string.
      Args:
        videoUrl: Full YouTube video URL.
        order: "relevance" (default) or "time".
        max: Max total comments to return (100–1000 建議).
      """

• server.py:15-31 (helper)

  Helper function to extract a YouTube video ID from various URL formats (youtube.com/watch, youtu.be, /embed/).

  def _extract_video_id(video_url: str) -> str:
      """Support https://www.youtube.com/watch?v=..., https://youtu.be/..., /embed/..."""
      try:
          u = urlparse(video_url)
          if u.netloc.endswith("youtu.be"):
              return u.path.strip("/")
          if "youtube.com" in u.netloc:
              qs = parse_qs(u.query or "")
              if "v" in qs:
                  return qs["v"][0]
              m = re.search(r"/embed/([A-Za-z0-9_-]{6,})", u.path or "")
              if m:
                  return m.group(1)
          m = re.search(r"([A-Za-z0-9_-]{11})", video_url)
          return m.group(1) if m else ""
      except Exception:
          return ""

• server.py:39-48 (helper)

  Helper function to extract and format a comment item from the YouTube API response into a simplified dictionary.

  def _pack_comment(item: Dict[str, Any], parent_id: Optional[str] = None) -> Dict[str, Any]:
      s = item.get("snippet", {})
      return {
          "id": item.get("id"),
          "parentId": parent_id,
          "author": s.get("authorDisplayName"),
          "publishedAt": s.get("publishedAt"),
          "likeCount": s.get("likeCount", 0),
          "text": s.get("textOriginal") or s.get("textDisplay") or "",
      }