import json
import logging
import anyio
from mcp.server.fastmcp import Context
from youtube_mcp_server.services.transcription_service import get_transcription_service
logger = logging.getLogger("youtube-mcp-server")
async def transcribe_video(url: str, ctx: Context, language: str = "auto") -> str:
"""
Transcribe a YouTube video.
Args:
url: YouTube video URL.
language: Target language code.
- 'auto': (Default) Detect language and transcribe in that language.
- 'en': Translate from source language to English.
- Other codes: Transcribe in the specified language.
Supported Language Codes:
af, am, ar, as, az, ba, be, bg, bn, bo, br, bs, ca, cs, cy, da, de, el, en, es, et, eu, fa, fi, fo, fr, gl, gu, ha, haw, he, hi, hr, ht, hu, hy, id, is, it, ja, jw, ka, kk, km, kn, ko, la, lb, ln, lo, lt, lv, mg, mi, mk, ml, mn, mr, ms, mt, my, ne, nl, nn, no, oc, pa, pl, ps, pt, ro, ru, sa, sd, si, sk, sl, sn, so, sq, sr, su, sv, sw, ta, te, tg, th, tk, tl, tr, tt, uk, ur, uz, vi, yi, yo, yue, zh
Examples:
- Transcribe untranslated (detected: French): `transcribe_video(url="...", language="auto")`
- Translate to English: `transcribe_video(url="...", language="en")`
- Transcribe specifically in German: `transcribe_video(url="...", language="de")`
Returns:
Unified JSON object with metadata and segments.
"""
await ctx.info(f"Starting transcription for: {url} (lang={language})")
import asyncio
loop = asyncio.get_running_loop()
def progress_callback(current, total, message):
# Bridge from worker thread to async event loop to send progress notification
try:
# anyio.from_thread can be finicky. Use standard asyncio threadsafe submission.
# We don't await the result here to avoid blocking, just fire and forget (or log error via callback if needed)
asyncio.run_coroutine_threadsafe(ctx.info(f"[{current}%] {message}"), loop)
# checking results of future is optional but good for debugging if needed,
# but here we just want non-blocking log dispatch.
except Exception as e:
# Log full representation to catch empty-message exceptions
logger.warning(f"Failed to send progress update: {repr(e)}")
try:
service = get_transcription_service()
# Run blocking service in a separate thread so we don't block the event loop
result = await anyio.to_thread.run_sync(
service.transcribe,
url,
language,
progress_callback
)
await ctx.info("Transcription completed")
return json.dumps(result, indent=2)
except Exception as e:
await ctx.error(f"Error transcribing video: {e}")
return f"Error transcribing video: {str(e)}"
