YouTube Transcript MCP Server

Instructions

Implementation Reference

• src/mcp_youtube_transcript/__init__.py:157-180 (handler)

  The handler function for the 'get_transcript' tool, registered via @mcp.tool() decorator. Handles fetching and paginating the transcript.

  @mcp.tool()
  async def get_transcript(
      ctx: Context[ServerSession, AppContext],
      url: str = Field(description="The URL of the YouTube video"),
      lang: str = Field(description="The preferred language for the transcript", default="en"),
      next_cursor: str | None = Field(description="Cursor to retrieve the next page of the transcript", default=None),
  ) -> Transcript:
      """Retrieves the transcript of a YouTube video."""
  
      title, snippets = _get_transcript_snippets(ctx.request_context.lifespan_context, _parse_video_id(url), lang)
      transcripts = (item.text for item in snippets)
  
      if response_limit is None or response_limit <= 0:
          return Transcript(title=title, transcript="\n".join(transcripts))
  
      res = ""
      cursor = None
      for i, line in islice(enumerate(transcripts), int(next_cursor or 0), None):
          if len(res) + len(line) + 1 > response_limit:
              cursor = str(i)
              break
          res += f"{line}\n"
  
      return Transcript(title=title, transcript=res[:-1], next_cursor=cursor)

• src/mcp_youtube_transcript/__init__.py:47-53 (schema)

  Pydantic model defining the output schema for the get_transcript tool.

  class Transcript(BaseModel):
      """Transcript of a YouTube video."""
  
      title: str = Field(description="Title of the video")
      transcript: str = Field(description="Transcript of the video")
      next_cursor: str | None = Field(description="Cursor to retrieve the next page of the transcript", default=None)

• src/mcp_youtube_transcript/__init__.py:109-124 (helper)

  Helper function to fetch the video title and transcript snippets using YouTubeTranscriptApi, with language fallback.

  @lru_cache
  def _get_transcript_snippets(ctx: AppContext, video_id: str, lang: str) -> Tuple[str, list[FetchedTranscriptSnippet]]:
      if lang == "en":
          languages = ["en"]
      else:
          languages = [lang, "en"]
  
      page = ctx.http_client.get(
          f"https://www.youtube.com/watch?v={video_id}", headers={"Accept-Language": ",".join(languages)}
      )
      page.raise_for_status()
      soup = BeautifulSoup(page.text, "html.parser")
      title = soup.title.string if soup.title and soup.title.string else "Transcript"
  
      transcripts = ctx.ytt_api.fetch(video_id, languages=languages)
      return title, transcripts.snippets

• src/mcp_youtube_transcript/__init__.py:98-107 (helper)

  Utility function to extract YouTube video ID from various URL formats.

  def _parse_video_id(url: str) -> str:
      parsed_url = urlparse(url)
      if parsed_url.hostname == "youtu.be":
          return parsed_url.path.lstrip("/")
      else:
          q = parse_qs(parsed_url.query).get("v")
          if q is None:
              raise ValueError(f"couldn't find a video ID from the provided URL: {url}.")
          return q[0]

• src/mcp_youtube_transcript/__init__.py:140-219 (registration)

  The server factory function where the get_transcript tool is registered by decorating its handler with @mcp.tool().

  def server(
      response_limit: int | None = None,
      webshare_proxy_username: str | None = None,
      webshare_proxy_password: str | None = None,
      http_proxy: str | None = None,
      https_proxy: str | None = None,
  ) -> FastMCP:
      """Initializes the MCP server."""
  
      proxy_config: ProxyConfig | None = None
      if webshare_proxy_username and webshare_proxy_password:
          proxy_config = WebshareProxyConfig(webshare_proxy_username, webshare_proxy_password)
      elif http_proxy or https_proxy:
          proxy_config = GenericProxyConfig(http_proxy, https_proxy)
  
      mcp = FastMCP("Youtube Transcript", lifespan=partial(_app_lifespan, proxy_config=proxy_config))
  
      @mcp.tool()
      async def get_transcript(
          ctx: Context[ServerSession, AppContext],
          url: str = Field(description="The URL of the YouTube video"),
          lang: str = Field(description="The preferred language for the transcript", default="en"),
          next_cursor: str | None = Field(description="Cursor to retrieve the next page of the transcript", default=None),
      ) -> Transcript:
          """Retrieves the transcript of a YouTube video."""
  
          title, snippets = _get_transcript_snippets(ctx.request_context.lifespan_context, _parse_video_id(url), lang)
          transcripts = (item.text for item in snippets)
  
          if response_limit is None or response_limit <= 0:
              return Transcript(title=title, transcript="\n".join(transcripts))
  
          res = ""
          cursor = None
          for i, line in islice(enumerate(transcripts), int(next_cursor or 0), None):
              if len(res) + len(line) + 1 > response_limit:
                  cursor = str(i)
                  break
              res += f"{line}\n"
  
          return Transcript(title=title, transcript=res[:-1], next_cursor=cursor)
  
      @mcp.tool()
      async def get_timed_transcript(
          ctx: Context[ServerSession, AppContext],
          url: str = Field(description="The URL of the YouTube video"),
          lang: str = Field(description="The preferred language for the transcript", default="en"),
          next_cursor: str | None = Field(description="Cursor to retrieve the next page of the transcript", default=None),
      ) -> TimedTranscript:
          """Retrieves the transcript of a YouTube video with timestamps."""
  
          title, snippets = _get_transcript_snippets(ctx.request_context.lifespan_context, _parse_video_id(url), lang)
  
          if response_limit is None or response_limit <= 0:
              return TimedTranscript(
                  title=title, snippets=[TranscriptSnippet.from_fetched_transcript_snippet(s) for s in snippets]
              )
  
          res = []
          size = len(title) + 1
          cursor = None
          for i, s in islice(enumerate(snippets), int(next_cursor or 0), None):
              snippet = TranscriptSnippet.from_fetched_transcript_snippet(s)
              if size + len(snippet) + 1 > response_limit:
                  cursor = str(i)
                  break
              res.append(snippet)
  
          return TimedTranscript(title=title, snippets=res, next_cursor=cursor)
  
      @mcp.tool()
      def get_video_info(
          ctx: Context[ServerSession, AppContext],
          url: str = Field(description="The URL of the YouTube video"),
      ) -> VideoInfo:
          """Retrieves the video information."""
          return _get_video_info(ctx.request_context.lifespan_context, url)
  
      return mcp