0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@zhangzhaoyu1204

【Python】YouTube文字起こしMCPサーバーを自作する｜MCP Python SDK v2（旧FastMCP）対応

0
Posted at

【Python】YouTube文字起こしMCPサーバーを自作する｜MCP Python SDK v2（旧FastMCP）対応

AIエージェントへYouTube URLを渡し、「この動画を文字起こしして」「処理状況を確認して」「結果から重要な発言を探して」と依頼できるMCPサーバーをPythonで作ります。

この記事では、MCP Python SDK v2を使って、非同期の動画文字起こしAPIを3つのMCPツールとして公開します。

create_transcription
  └─ YouTube URLから文字起こしタスクを作成

get_transcription_status
  └─ request_idから処理状況を確認

get_transcription_result
  └─ 完了した文字起こしとタイムスタンプを取得

2026年9月時点のMCP Python SDKでは、v2が安定版です。v1で使われていたFastMCPクラスは、v2ではMCPServerへ改名されています。

そのため本記事は、古いfrom mcp.server.fastmcp import FastMCPではなく、v2の次の書き方を使用します。

from mcp.server import MCPServer

この記事で作るYouTube文字起こしMCPサーバー

全体構成は次のとおりです。

MCPホスト
   ↓ ツール呼び出し
Python MCPサーバー
   ↓ Bearer認証
Transcript API
   ↓
YouTube動画の文字起こし結果

MCPは、LLMやAIエージェントから外部のデータや機能を利用するための標準的なインターフェースです。

今回は、YouTube URLなどの公開メディアを非同期処理できるYouTubeや動画URLをテキスト化できるTranscript APIを、MCPツールの背後で呼び出します。

なぜ一つのMCPツールで完了まで待たないのか

最初は次のようなツールを考えたくなります。

transcribe_youtube(url)
  → 文字起こしが終わるまで待つ
  → 全文を返す

しかし、長時間動画ではツール呼び出しが長くなり、MCPホスト側のタイムアウトや再実行につながります。再実行されると、同じ文字起こしタスクを重複して作る可能性もあります。

そこで今回は、外部APIの非同期ライフサイクルをMCP側にも反映します。

1. タスクを作る
2. 指定された時間を待つ
3. 状態を確認する
4. 成功したら結果を取得する

AIエージェントが状態を理解できるよう、各ツールはrequest_id、status、retry_afterなどの構造化データを返します。

動作環境

  • Python 3.10以上
  • MCP Python SDK v2
  • httpx
  • python-dotenv
  • pydantic

uvを使ってプロジェクトを作成します。

uv init youtube-transcript-mcp
cd youtube-transcript-mcp
uv add "mcp[cli]" httpx python-dotenv pydantic

pipを使う場合は次のとおりです。

python -m venv .venv

# macOS / Linux
source .venv/bin/activate

# Windows PowerShell
.venv\Scripts\Activate.ps1

pip install "mcp[cli]" httpx python-dotenv pydantic

ファイル構成はシンプルです。

youtube-transcript-mcp/
├── .env
├── .gitignore
├── pyproject.toml
└── server.py

.envへAPI Keyを保存します。

VT_API_KEY=your_api_key

.env
.venv/
__pycache__/

API KeyはMCPツールの引数にせず、サーバー側の環境変数から読み込みます。ツール引数にすると、MCPホストの会話履歴やツール実行ログへ秘密情報が残る可能性があります。

MCP Python SDK v2でサーバーを作る

まずはサーバー本体を作成します。

# server.py
from __future__ import annotations

import hashlib
import os
import re
from typing import Any
from urllib.parse import urlparse

import httpx
from dotenv import load_dotenv
from mcp.server import MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from pydantic import BaseModel, Field

load_dotenv()

API_BASE = "https://videotranscriber.ai/openapi/v1"
YOUTUBE_HOSTS = {
    "youtube.com",
    "www.youtube.com",
    "m.youtube.com",
    "youtu.be",
}

mcp = MCPServer(
    "youtube-transcript",
    title="YouTube Transcript MCP Server",
    description="Create and retrieve timestamped YouTube transcripts.",
    instructions=(
        "Use create_transcription first. "
        "Wait for retry_after seconds before calling "
        "get_transcription_status. "
        "Call get_transcription_result only after the task succeeds."
    ),
    version="1.0.0",
)

v2では、name以外はキーワード引数で渡すのが安全です。サーバー名、説明、バージョンを明示しておくと、MCPホストのツール一覧やログでも識別しやすくなります。

MCPツールの戻り値をPydanticで定義する

辞書をそのまま返すこともできますが、戻り値の型を定義すると、MCPホストが構造を理解しやすくなります。

class CreateTaskResult(BaseModel):
    request_id: str
    status: str
    retry_after: float = Field(ge=1)


class TaskStatusResult(BaseModel):
    request_id: str
    status: str
    retry_after: float | None = None
    error_code: str | None = None
    retryable: bool | None = None


class TranscriptSegment(BaseModel):
    start: float
    end: float
    text: str
    speaker: str | None = None


class TranscriptResult(BaseModel):
    request_id: str
    language: str | None = None
    duration: float | None = None
    full_text: str | None = None
    segments: list[TranscriptSegment]
    segment_offset: int
    next_offset: int | None = None
    total_segments: int

get_transcription_resultでは、全文を毎回返さない設計にしています。長い動画の全文を一度にMCPへ返すと、モデルのコンテキストを大量に消費するためです。

YouTube URLとrequest_idを検証する

MCPツールへ文字列引数を渡せるからといって、受け取った値をそのまま外部APIへ送るべきではありません。

def validate_youtube_url(source_url: str) -> str:
    try:
        parsed = urlparse(source_url)
    except ValueError as exc:
        raise ToolError("Invalid YouTube URL") from exc

    hostname = (parsed.hostname or "").lower()
    if parsed.scheme != "https" or hostname not in YOUTUBE_HOSTS:
        raise ToolError(
            "source_url must be a public HTTPS YouTube URL"
        )

    return source_url


def validate_request_id(request_id: str) -> str:
    if not re.fullmatch(r"tr_[A-Za-z0-9_-]+", request_id):
        raise ToolError("Invalid transcription request_id")
    return request_id

今回はYouTube文字起こしMCPサーバーなので、ホスト名をYouTubeに限定します。任意URLを許可する場合は、SSRF対策としてプライベートIP、認証情報付きURL、リダイレクト先なども検証する必要があります。

Transcript APIを呼ぶ共通関数

次に、Bearer認証とエラー処理を共通化します。

def api_key() -> str:
    value = os.environ.get("VT_API_KEY")
    if not value:
        raise ToolError("VT_API_KEY is not configured")
    return value


async def api_request(
    method: str,
    path: str,
    **kwargs: Any,
) -> tuple[dict[str, Any], httpx.Response]:
    headers = {
        "Authorization": f"Bearer {api_key()}",
        **kwargs.pop("headers", {}),
    }

    try:
        async with httpx.AsyncClient(
            base_url=API_BASE,
            headers=headers,
            timeout=httpx.Timeout(30.0),
        ) as client:
            response = await client.request(
                method,
                path,
                **kwargs,
            )
    except httpx.TimeoutException as exc:
        raise ToolError("Transcript API timed out") from exc
    except httpx.TransportError as exc:
        raise ToolError("Could not connect to Transcript API") from exc

    try:
        body = response.json()
    except ValueError as exc:
        raise ToolError(
            f"Transcript API returned non-JSON: HTTP {response.status_code}"
        ) from exc

    if response.is_error:
        error = body.get("error", {})
        code = error.get("code", "unknown_error")
        retryable = error.get("retryable", False)

        # message全文や認証情報をそのまま返さない
        raise ToolError(
            f"Transcript API error: status={response.status_code}, "
            f"code={code}, retryable={retryable}"
        )

    return body, response

エラー処理で重要なのは、APIの内部メッセージやリクエストヘッダーをそのままモデルへ返さないことです。AIエージェントが判断に必要なHTTPステータス、安定したエラーコード、再試行可能かどうかだけを返します。

Tool 1：文字起こしタスクを作成する

一つ目のMCPツールは、YouTube URLから非同期タスクを作成します。

@mcp.tool()
async def create_transcription(
    source_url: str,
    language: str = "auto",
    speaker_diarization: bool = False,
    operation_id: str | None = None,
) -> CreateTaskResult:
    """Create a transcription task for a public YouTube URL.

    Call this tool once for one logical transcription operation.
    Reuse operation_id when retrying the same operation.
    """
    source_url = validate_youtube_url(source_url)

    if operation_id:
        if not re.fullmatch(r"[A-Za-z0-9._:-]{8,100}", operation_id):
            raise ToolError("Invalid operation_id")
        idempotency_key = f"mcp-{operation_id}"
    else:
        material = (
            f"{source_url}|{language}|{speaker_diarization}"
        ).encode("utf-8")
        digest = hashlib.sha256(material).hexdigest()[:32]
        idempotency_key = f"mcp-{digest}"

    body, _ = await api_request(
        "POST",
        "/transcriptions",
        headers={"Idempotency-Key": idempotency_key},
        json={
            "source_url": source_url,
            "language": language,
            "speaker_diarization": speaker_diarization,
        },
    )

    return CreateTaskResult(
        request_id=body["request_id"],
        status=body["status"],
        retry_after=max(float(body.get("retry_after", 5)), 1.0),
    )

operation_idを省略した場合は、URLとオプションから安定したIdempotency-Keyを生成します。同じ条件でツールが再実行されても、別のタスクとして扱われにくくなります。

意図的に新しいタスクを作成したい場合は、新しいoperation_idを渡します。

Tool 2：文字起こしの状態を確認する

二つ目のツールはrequest_idから処理状態を取得します。

@mcp.tool()
async def get_transcription_status(
    request_id: str,
) -> TaskStatusResult:
    """Get the current status of a transcription task.

    If the task is queued or processing, wait for retry_after seconds
    before calling this tool again.
    """
    request_id = validate_request_id(request_id)
    body, response = await api_request(
        "GET",
        f"/transcriptions/{request_id}",
    )

    retry_after_value = (
        body.get("retry_after")
        or response.headers.get("Retry-After")
    )

    error = body.get("error") or {}

    return TaskStatusResult(
        request_id=request_id,
        status=body["status"],
        retry_after=(
            max(float(retry_after_value), 1.0)
            if retry_after_value is not None
            else None
        ),
        error_code=error.get("code"),
        retryable=error.get("retryable"),
    )

ツールのdocstringに「retry_after秒待ってから再実行する」と書いておくことで、MCPホストがツールの意図を理解しやすくなります。

状態には、主に次の値があります。

queued
processing
succeeded
partial_succeeded
failed
cancelled

succeededまたはpartial_succeededになったら、結果取得ツールを呼びます。

Tool 3：文字起こし結果を取得する

長い文字起こしを安全に扱うため、セグメントをoffsetとlimitで分割して返します。

@mcp.tool()
async def get_transcription_result(
    request_id: str,
    segment_offset: int = 0,
    segment_limit: int = 100,
    include_full_text: bool = False,
) -> TranscriptResult:
    """Get a completed timestamped transcript.

    Use segment_offset and segment_limit to avoid returning an entire
    long transcript in one tool response.
    """
    request_id = validate_request_id(request_id)

    if segment_offset < 0:
        raise ToolError("segment_offset must be >= 0")
    if not 1 <= segment_limit <= 200:
        raise ToolError("segment_limit must be between 1 and 200")

    body, _ = await api_request(
        "GET",
        f"/transcriptions/{request_id}/result",
    )

    transcript = body.get("transcript", body)
    raw_segments = transcript.get("segments", [])
    selected = raw_segments[
        segment_offset:segment_offset + segment_limit
    ]

    segments = [
        TranscriptSegment(
            start=float(item["start"]),
            end=float(item["end"]),
            text=str(item.get("text", "")).strip(),
            speaker=item.get("speaker"),
        )
        for item in selected
        if str(item.get("text", "")).strip()
    ]

    next_offset = segment_offset + len(selected)
    if next_offset >= len(raw_segments):
        next_offset = None

    return TranscriptResult(
        request_id=request_id,
        language=transcript.get("language")
        or transcript.get("detected_language"),
        duration=(
            float(transcript["duration"])
            if transcript.get("duration") is not None
            else None
        ),
        full_text=(
            transcript.get("text")
            or transcript.get("full_text")
            if include_full_text
            else None
        ),
        segments=segments,
        segment_offset=segment_offset,
        next_offset=next_offset,
        total_segments=len(raw_segments),
    )

segment_limitの最大値を200に制限しています。動画が長い場合は、返されたnext_offsetを次の呼び出しへ渡します。

1回目：segment_offset=0
2回目：segment_offset=100
3回目：segment_offset=200

全文が本当に必要な場合だけ、include_full_text=Trueを指定します。

サーバーをstdioで起動する

最後に、ファイル末尾へ起動処理を追加します。

if __name__ == "__main__":
    mcp.run()

引数なしのmcp.run()は、ローカルMCPサーバーで一般的なstdioトランスポートを使用します。

run()を必ずif __name__ == "__main__":の中へ置きます。MCP Inspectorやテストコードがserver.pyをインポートしただけでサーバーが起動するのを防ぐためです。

MCP Inspectorでテストする

MCP Python SDKのCLIを使うと、Inspectorからツールを確認できます。

uv run mcp dev server.py

InspectorのTools画面に、次の3ツールが表示されれば準備完了です。

create_transcription
get_transcription_status
get_transcription_result

最初にcreate_transcriptionを実行します。

{
  "source_url": "https://www.youtube.com/watch?v=VIDEO_ID",
  "language": "auto",
  "speaker_diarization": false,
  "operation_id": "qiita-demo-001"
}

レスポンス例です。

{
  "request_id": "tr_01JEXAMPLE",
  "status": "queued",
  "retry_after": 5
}

指定秒数を待ってから、get_transcription_statusを実行します。

{
  "request_id": "tr_01JEXAMPLE"
}

成功後、get_transcription_resultを実行します。

{
  "request_id": "tr_01JEXAMPLE",
  "segment_offset": 0,
  "segment_limit": 100,
  "include_full_text": false
}

MCPホストへ登録する

ローカルのstdioサーバーは、MCPホストからコマンドとして起動します。設定形式や保存場所はホストごとに異なりますが、概念的には次のような設定になります。

{
  "mcpServers": {
    "youtube-transcript": {
      "command": "uv",
      "args": [
        "run",
        "--directory",
        "/absolute/path/to/youtube-transcript-mcp",
        "python",
        "server.py"
      ],
      "env": {
        "VT_API_KEY": "your_api_key"
      }
    }
  }
}

API Keyを設定ファイルへ直接書きたくない場合は、MCPホストが提供するシークレット管理や、OSの環境変数を利用してください。

登録後は、例えば次のように依頼できます。

このYouTube動画の文字起こしタスクを作成してください。
完了したら、10分から15分までの発言を取得してください。

ただし、MCPホストが自動的に待機や再呼び出しを行うかは実装によって異なります。長時間タスクを完全自動化する場合は、サーバー側へジョブ管理や通知機構を追加します。

FastMCP v1の記事をv2へ読み替える

既存の記事には次のコードが多くあります。

# MCP Python SDK v1
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("youtube-transcript")

v2では次のように変更します。

# MCP Python SDK v2
from mcp.server import MCPServer

mcp = MCPServer("youtube-transcript")

日常的に使うデコレーターは同じです。

@mcp.tool()
async def example_tool(value: str) -> dict:
    return {"value": value}

FastMCPという検索キーワードで古い記事へたどり着いた場合は、インストールされているMCP Python SDKのメジャーバージョンを先に確認します。

python -c "import importlib.metadata; print(importlib.metadata.version('mcp'))"

v1とv2のコードを同じファイルで混在させると、ImportErrorや型の不一致が起きやすいため注意してください。

MCPツール設計で重要だったポイント

1. 外部APIの状態を隠しすぎない

非同期APIを無理に同期ツールへ見せると、タイムアウト時に何が起きたか分からなくなります。request_idとstatusをそのままMCPの構造化データへ含めます。

2. docstringを操作説明として書く

MCPツールのdocstringは、人間向けコメントだけではありません。ツールを選択するAIエージェントが、いつ呼ぶべきかを判断する材料になります。

「何をするか」だけでなく、次にどのツールを呼ぶか、何秒待つべきかも説明します。

3. 大きなレスポンスを一度に返さない

1時間以上の動画文字起こしを一度にモデルへ返すと、コンテキストを圧迫します。セグメント単位で分割し、必要な範囲だけ返します。

4. エラーをモデルが判断できる形にする

単にExceptionの文字列を返すのではなく、再試行可能か、入力修正が必要かを判断できる情報へ絞ります。

5. API Keyをツール引数にしない

ツール引数と戻り値はログに残る前提で設計します。API Key、Authorizationヘッダー、署名付きダウンロードURLなどは、モデルへ見せないようにします。

よくあるエラー

ModuleNotFoundError: mcp.server.fastmcp

MCP Python SDK v2ではFastMCPのimport pathが変更されています。

from mcp.server import MCPServer

へ変更し、v1とv2のコードが混ざっていないか確認します。

MCP Inspectorにツールが表示されない

  • mcpがモジュール直下で作成されているか
  • @mcp.tool()が付いているか
  • 関数の型注釈が正しいか
  • mcp.run()がmain guard内にあるか

を確認します。

同じタスクが複数作成される

同じ論理操作を再試行するときに、別のoperation_idを渡していないか確認します。作成リクエストの再送では、同じIdempotency-Keyを使う必要があります。

文字起こし結果が大きすぎる

include_full_text=Falseにし、segment_limitを小さくします。返されたnext_offsetを使って必要な範囲だけ取得します。

stdioサーバーが何も表示しない

python server.pyを直接実行した場合、stdioサーバーはMCPホストからの入力を待つため、何も表示せず停止したように見えます。Inspector経由で接続して確認します。

Streamable HTTPで公開する場合

ローカル利用ではstdioが簡単ですが、リモート環境へデプロイする場合はStreamable HTTPを使えます。

if __name__ == "__main__":
    mcp.run(
        transport="streamable-http",
        host="127.0.0.1",
        port=8000,
    )

接続先はデフォルトで次の形式です。

http://127.0.0.1:8000/mcp

インターネットへ公開する場合は、TLS、認証、アクセス制御、レート制限、監査ログが必要です。API Keyを持つMCPサーバーを認証なしで公開しないでください。

また、新規実装では旧SSEではなくStreamable HTTPを選びます。

まとめ

PythonでYouTube文字起こしMCPサーバーを作ると、AIエージェントから動画処理APIを構造化されたツールとして利用できます。

今回のポイントは次の5つです。

  1. MCP Python SDK v2のMCPServerを使用する
  2. タスク作成・状態確認・結果取得を別ツールにする
  3. Idempotency-Keyで重複作成を防ぐ
  4. 長い文字起こしを分割して返す
  5. API Keyや内部エラーをモデルへ渡さない

既存のFastMCP入門記事の多くは、足し算や天気APIを例にしています。非同期の文字起こしAPIを題材にすると、タスク状態、再実行、レスポンスサイズ、秘密情報管理など、実際のMCPツール開発で必要になる設計をまとめて確認できます。

まずはstdioとMCP Inspectorで3つのツールをテストし、必要になった段階でStreamable HTTPや永続ジョブ管理へ拡張するのがおすすめです。

※MCP SDK、APIの仕様、利用制限、料金は変更される場合があります。実装時には最新の公式ドキュメントを確認してください。また、動画の利用条件と著作権を確認し、処理する権限のあるコンテンツで使用してください。

Qiita投稿時のタグ

Python MCP MCPサーバー YouTube 文字起こし

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?