-
Notifications
You must be signed in to change notification settings - Fork 1.5k
Description
I'm trying to make mcp for using youtube-transcript-api, but when I tried to run it and send message it get error ,
McpError: Connection closed
I manually test my mcp and it seems working
here is code for my mcp and the agent
---------------MCP------------------------------------------------------------------
from mcp.server.fastmcp import FastMCP
from youtube_transcript_api import YouTubeTranscriptApi
import requests
import re
mcp = FastMCP("YouTube Transcript Tool")
NAME: YouTube Transcript Fetcher + Saver
def id_from_url(url):
if 'youtu.be' in url:
return url.split('/')[-1]
return url.split("?v=")[-1].split("&")[0]
def get_title_and_author(video_id):
try:
oembed_url = f"https://www.youtube.com/oembed?url=https://www.youtube.com/watch?v={video_id}&format=json"
response = requests.get(oembed_url)
response.raise_for_status()
data = response.json()
return data.get('title', 'Unknown Title'), data.get('author_name', 'Unknown Author')
except:
return 'Unknown Title', 'Unknown Author'
def format_time(seconds):
hours = int(seconds // 3600)
minutes = int((seconds % 3600) // 60)
seconds = int(seconds % 60)
return f"{hours:02}:{minutes:02}:{seconds:02}"
def format_transcript(transcript_items):
formatted_segments = []
segment_start = None
segment_texts = []
for item in transcript_items:
start = item.start
text = item.text
if segment_start is None or (start - segment_start >= 30):
if segment_texts:
start_time = format_time(segment_start)
formatted_segments.append(f"({start_time}) {' '.join(segment_texts)}")
segment_texts = []
segment_start = start
segment_texts.append(text)
if segment_texts:
start_time = format_time(segment_start)
formatted_segments.append(f"({start_time}) {' '.join(segment_texts)}")
return "\n".join(formatted_segments)
@mcp.tool()
def fetch_youtube_transcript(video_url: str) -> str:
"""
Fetch and save the transcript of a YouTube video. Returns transcript text and path to saved file.
Args:
video_url: YouTube url to get transcript
Returns:
String for transcript of the video
"""
try:
video_id = id_from_url(video_url)
title, author = get_title_and_author(video_id)
transcript_list = YouTubeTranscriptApi.list_transcripts(video_id)
transcript = transcript_list.find_transcript(['en'])
transcript_items = transcript.fetch()
formatted = format_transcript(transcript_items)
full_text = f"{title} by {author}\n{video_url}\n\n{formatted}"
return f"Transcript {full_text}"
except Exception as e:
return f"Error processing transcript: {str(e)}"
if name == "main":
mcp.run(transport="stdio")
---------------END MCP------------------------------------------------------------------
--------------AGENT---------------------------
from google.adk.agents import Agent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters,StdioConnectionParams
import os
import sys
from dotenv import load_dotenv
load_dotenv()
os.environ["GOOGLE_API_KEY"] = os.getenv("GOOGLE_API_KEY")
MODEL_ID = "gemini-2.5-flash-preview-04-17"
PROMPT = 'Summarize this video into the key sections, ideas and action items. Deliver the notes from a first person perspective. If there are slides or info graphics in the video then include a description of them with any text that appears on there. Also make a transcript from the video after summarize sa as md file and filename will be date today in DDMMYYYY format'
tools = MCPToolset(
connection_params=StdioConnectionParams(
timeout=15,
server_params=StdioServerParameters(
command=sys.executable,
args=['youtube_summarizer/yt_transcript.py'],
)
)
)
root_agent = Agent(
model=MODEL_ID, # Adjust if needed
name='youtube_summarizer',
instruction=PROMPT,
tools=[tools]
)
-------------END AGENT --------------------------