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