ClaudeのMCPを徹底解説！ & gpt-4o+MCP+YouTube APIの動画推薦チャットAIも作る

こんにちは！逆瀬川 ( https://x.com/gyakuse ) です！

このアドベントカレンダーでは生成AIのアプリケーションを実際に作り、どのように作ればいいのか、ということをわかりやすく書いていければと思います。アプリケーションだけではなく、プロダクト開発に必要なモデルの調査方法、training方法、基礎知識等にも触れていければと思います。アドベントカレンダー全部書く、みたいな予定を立ててしまったので、1日あたり60分くらいで書けたらな〜と思っていますが、以下の記事は3時間程度かかり、泣いています。勢いで書いているので、不正確な部分がもしあればコメント等で指摘していただければ幸いです。

## 今回の記事について

今日はClaudeが発表したMCP ([Model Context Protocol](https://modelcontextprotocol.io/)) について紹介し、独自のMCPサーバーを作成したり、Claude以外のLLMで使えることを見せていこうと思います。MCPを取り上げたのは、最近注目されているエージェントシステムを作る上で重要な概念が盛り込まれているためです。

## 今日の記事で得られる知識

- MCPについて
- 具体的な実装方法

今日の記事を読むと、MCP+ChatGPT+YouTubeの動画推薦アプリが作れます。

![Cursor_と_Gradio.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2958603/d12199bd-70bd-96b6-a35c-4d55e170fede.png)

## MCPの前に

さて、MCPとはなんぞや、という前に、AIエージェントについて考えてみましょう。

MCPは実はかなりシンプルなプロトコルなので (それ故に、みんな作っていると思います。MCPが今後一般化するかどうかはかなり不明瞭です)、AIエージェントについて考えると、必然的に要求されるものではあるのです。

![ai_tool_svg.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2958603/f9b6caf5-dbd3-44c4-8214-3f4b6a91622a.png)

例えば天気について特化したAIエージェントは、天気に関するさまざまなツールについて理解して、ユーザーの命令 (「今日の天気を教えて」等)に対して、適切なツールを選び、ツールが実行され、得られた情報をもとに解答することができます。

この話をすると、おそらく関数呼び出し (Function Calling) について思い出されたかたもいるのではないでしょうか。関数呼び出しはユーザーの自然言語による指示から適切なツールを選ぶことができるものです。MCPは、そもそもツールについて、ちゃんと定義しないとね、みたいなモチベーションから必要とされています。

関数呼び出しの定義は関数名、説明、必要なパラメータ、期待される返り値の集合でした。天気予報に関する関数呼び出しであれば、場所、日付がパラメータとして、予報結果情報が期待される返り値として定義されます。しかし、この関数呼び出しは実際の天気情報APIを叩いたりするところをカバーしません。ゆえに、ツールについての定義が必要なのです。では、どういうふうに定義すればいいでしょうか？ (訂正: 多くの関数呼び出しは関数自体の返り値は保証しない。関数と必要なパラメータの定義が主)

天気情報についてまとめたツールサーバーには以下のようなことをしてくれるとよいです。

- サーバーに繋がると、以下を教えてくれる
    - 利用可能なツールリスト
        - ツールごとにパラメータや返り値の定義がされている
    - ツールを組み合わせていい感じにするワークフロー的なもの
- ツールやワークフローに適切なリクエストを送ると実行して結果がいい感じに返ってくる
    - 天気だったら、テキストのほかに画像とかもいい感じに帰ってきてほしい

こういう、ツールサーバーこうあってほしいな、みたいなのをプロトコルとしてまとめたのがMCPです。

## MCPとは

> [Model Context Protocol](https://modelcontextprotocol.io/)（MCP）は、LLMアプリケーションと外部データソースやツールとの間でシームレスな統合を可能にするオープンプロトコルです。AIを活用したIDEの構築、チャットインターフェースの強化、カスタムAIワークフローの作成など、MCPはLLMに必要なコンテキストを提供するための標準化された接続方法を提供します。
> MCPはオープンなプロトコルであり、どのアプリケーションにも統合可能です。IDEs、AIツール、その他のソフトウェアがMCPを使用してローカル統合に標準化された方法で接続できます。
https://modelcontextprotocol.io/introduction より翻訳して引用

上記の説明だとちょっとわかりづらいかもしれませんが、さきほど説明したように、LLMに対して各種ツールを提供するための仕組みと思って大丈夫です。

### MCPに登場する用語

- **MCP Hosts (MCPホスト)**: Claude Desktop、IDEs、AIツールなど、MCPを通じてリソースにアクセスしたいプログラム
    - これはつまりユーザーとの対話のフロント部分です
- **MCP Clients (MCPクライアント)**: サーバーと1対1の接続を維持するプロトコルクライアント
    - これはホストとサーバーの接続を仲介するためのサービス層として捉えればよいです
- **MCP Servers (MCPサーバー)**: 各サーバーが標準化されたModel Context Protocolを介して特定の機能を提供する軽量プログラム
    - これがサーバー実態です
- **Local Resources (ローカルリソース)**: コンピュータのデータベース、ファイル、サービスなど、MCPサーバーが安全にアクセスできるリソース
    - サーバーが扱うリソースです
- **Remote Resources (リモートリソース)**: APIを介してMCPサーバーが接続するインターネット上のリソース
    - クラウド上にあるリソースです

## MCPサーバー

では次にMCPサーバーについて詳細に見ていきます。

MCPサーバーは以下を提供します。

- [リソース](https://modelcontextprotocol.io/docs/concepts/resources)
    - ログ、画像、(MCPサーバーが例えば外部APIと通信するものだったら) APIレスポンス、等々
- [プロンプト](https://modelcontextprotocol.io/docs/concepts/prompts)
    - 対話的にワークフロー (単一または複数のツール実行) を行うための機構
    - MCPクライアントからはツールが直接呼ばれるケースとプロンプトで呼ばれるケースの2つがあります。
- [ツール](https://modelcontextprotocol.io/docs/concepts/tools)
    - 実行の実態です。
    - `tools/list` でMCPクライアントからはツール群を取得できます
- 通知機能
    - サーバーにリソース、プロンプト、ツール等が追加されたときに通知します
    - リソースをwatchしておけば、いまここ、みたいな情報を出せます、便利。

## MCPサーバーの実装例

### ツールの実装

以下、実装サンプルです。`call_tool` の `if_name` 以下に実装を書き、`list_tools` に追加するだけの簡単なお仕事です。

```python
app = Server("example-server")

@app.list_tools()
async def list_tools() -> list[types.Tool]:
    return [
        types.Tool(
            name="calculate_sum",
            description="Add two numbers together",
            inputSchema={
                "type": "object",
                "properties": {
                    "a": {"type": "number"},
                    "b": {"type": "number"}
                },
                "required": ["a", "b"]
            }
        )
    ]

@app.call_tool()
async def call_tool(
    name: str,
    arguments: dict
) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
    if name == "calculate_sum":
        a = arguments["a"]
        b = arguments["b"]
        result = a + b
        return [types.TextContent(type="text", text=str(result))]
    raise ValueError(f"Tool not found: {name}")

```

## SQLite MCPサーバーの具体例

![mcp_svg.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2958603/6a299a0a-db94-fcf4-e970-b0814477e757.png)

より良く理解するためにSQLite MCPサーバーについて考えましょう。

このSQLiteサーバーには店舗の売上データが入っています。

**事前段階**

- MCPサーバーとハンドシェイクし、`tools/list` 等を得ます。
- 動的に `tools/list` や `prompts/list` を関数呼び出しに変換してもいいかもしれません

**利用段階**

ユーザーが「今売れてる商品教えて」とリクエストした場合の流れを考えます。

1. MCPホストでのツール検出
    - ホストアプリケーションがリクエスト内容を解析します
    - Function Calling
        - リクエストが「売上トップの商品を取得する」操作に対応していると判断します
        - ツール`top-selling`が適用されるべきと推測されます
2. パラメータ検証
    - ツール`top-selling`には、以下の引数が必要とします
        - `date`: 売上データの日付（必須）
        - `limit`: トップ商品の数 (オプション、デフォルトは5)
    - MCPホストは、ユーザーリクエストや会話履歴からこれらの引数を取得または補完します。
        - 例: `date`が指定されていない場合、「今日の日付」をデフォルト値として設定。
3. SQL文の生成と送信
    - MCPホストは適切なSQL文を生成します
    - このSQL文を、ツール`top-selling`に対応するMCPエンドポイントに送信します
    - MCPサーバーのエンドポイントは `tools/call` となります
4. MCPサーバーでの処理
    - サーバーは`tools/call`リクエストを受信します
        - ツール名: `top-selling`
        - 引数: `{"date": "2024-12-01", "limit": 5}`
    - サーバーはリクエストを検証し、対応するSQL文を実行します。
    - サーバー内部でSQLiteデータベースをクエリして結果を取得します。
5. 結果をホストに返却
    - MCPサーバーはクエリ結果をJSON形式でホストに返却します:
6. ホストでの結果処理
    - MCPホストは受け取ったデータを整形し、ユーザーに自然言語で返答します。
    - 例: 「今日の売上トップ商品は以下の通りです: 1位 りんご (5000円), 2位 ごりら (4500円), 3位 ラッパ (3000円)...」

と、こんな感じでMCPサーバーは作成されます

## Claude以外のLLM + MCPサーバー

最後に実際に適当なMCPサーバーを作り、Claude以外のLLMを用いて動かすやつをやっていきましょう。function callingでうまくいくかは謎です。いったらいいなぁ。とりあえず実装していきましょう。

### MCPサーバーを作る

https://modelcontextprotocol.io/docs/first-server/python

https://github.com/modelcontextprotocol/create-python-server

今回は[YouTube Data API v3](https://console.cloud.google.com/apis/api/youtube.googleapis.com/)を使って、YouTubeの検索をしてくれるツールを作ります。

```bash
uvx create-mcp-server
```

`mcp_server_youtube` という名前にしました。

- `mcp_server_youtube` というディレクトリができます。
- `mcp_server_youtube/src/mcp_server_youtube/server.py` にサーバー実装を記述します。

**実装**

MCPサーバーの実装はほとんどgpt-4oを使って行いました。

- ポイント
    - 今回はこのサーバーに登録されたツールが `youtube-search` のみなので、`handle_call_tool` に到着したリクエストが `youtube-search` と一致している場合のみ処理行います
    - YouTube Data API v3 は単純にAPIを実装するだけです
- これの嬉しさ
    - 普通にfunction callingからAPIを叩くだけなら、MCPサーバーはいりません。ただ、独立したMCPサーバーとして作ることで再利用がしやすい形になります

```python
import asyncio
from googleapiclient.discovery import build
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
import mcp.server.stdio
from pydantic import AnyUrl
import os
from dotenv import load_dotenv

# 環境変数を読み込み
load_dotenv()

# YouTube Data APIキーを設定
YOUTUBE_API_KEY = os.getenv("YOUTUBE_API_KEY")
if not YOUTUBE_API_KEY:
    raise ValueError("YouTube API Key is not set in the environment variables.")
youtube = build("youtube", "v3", developerKey=YOUTUBE_API_KEY)

# MCPサーバーの初期化
server = Server("mcp_server_youtube")

# 動画検索結果を保持するリソース（サーバーの状態）
search_results = {}

# YouTube Data APIを利用して動画を検索する関数
def search_youtube_videos(query: str, max_results: int = 5):
    """
    YouTube動画を検索して結果を返す。
    """
    request = youtube.search().list(
        q=query,
        part="snippet",
        type="video",
        maxResults=max_results
    )
    response = request.execute()

    # 動画情報を整形して返す
    videos = []
    for item in response["items"]:
        video_info = {
            "title": item["snippet"]["title"],
            "description": item["snippet"]["description"],
            "channel": item["snippet"]["channelTitle"],
            "url": f"https://www.youtube.com/watch?v={item['id']['videoId']}"
        }
        videos.append(video_info)
    return videos

# ツールをリストするエンドポイント
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
    """
    ツールのリストを提供する。
    """
    return [
        types.Tool(
            name="youtube-search",
            description="Search for YouTube videos by query.",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "The search query for YouTube videos."
                    },
                    "max_results": {
                        "type": "integer",
                        "description": "Maximum number of results to return.",
                        "default": 5
                    }
                },
                "required": ["query"]
            }
        )
    ]

# ツールを実行するエンドポイント
@server.call_tool()
async def handle_call_tool(
    name: str, arguments: dict
) -> list[types.TextContent | types.EmbeddedResource]:
    """
    ツールを実行してYouTube検索を行う。
    """
    if name != "youtube-search":
        raise ValueError(f"Unknown tool: {name}")

    # 引数を取得
    query = arguments.get("query")
    max_results = arguments.get("max_results", 5)

    # YouTube検索を実行
    videos = search_youtube_videos(query, max_results)

    # 結果をリソースに保存
    search_results[query] = videos

    # クライアントに結果を返す
    results_text = "\n".join(
        [f"{video['title']} ({video['url']}) - {video['channel']}" for video in videos]
    )
    return [
        types.TextContent(
            type="text",
            text=f"Search results for '{query}':\n{results_text}"
        )
    ]

# リソースをリストするエンドポイント
@server.list_resources()
async def handle_list_resources() -> list[types.Resource]:
    """
    検索結果をリソースとしてリストする。
    """
    return [
        types.Resource(
            uri=AnyUrl(f"youtube-search://{query}"),
            name=f"Search Results for '{query}'",
            description=f"Search results for query: '{query}'",
            mimeType="text/plain",
        )
        for query in search_results
    ]

# リソースを読み取るエンドポイント
@server.read_resource()
async def handle_read_resource(uri: AnyUrl) -> str:
    """
    指定された検索クエリの結果を返す。
    """
    query = uri.path.lstrip("/")
    if query not in search_results:
        raise ValueError(f"No results found for query: {query}")

    # 結果を整形して返す
    videos = search_results[query]
    return "\n".join(
        [f"{video['title']} ({video['url']}) - {video['channel']}" for video in videos]
    )

# メイン関数
async def main():
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="mcp_server_youtube",
                server_version="0.0.1",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={}
                )
            )
        )

# 実行
if __name__ == "__main__":
    asyncio.run(main())
```

### ホスト側を実装する

では次にホスト側をgradioで実装していきます。
これも `gpt-4o` にサポートしてもらいつつ実装しました。

- ポイント
    - gpt-4oのFunction Callingを用いて意図検出 (MCPホストとしての役割)
    - MCPクライアント部分は `mcp_search_youtube` で行っている
- もうちょっと頑張る場合
    - subscribeするMCPサーバーリストが与えられたら、起動時にハンドシェイクを行い、`tools/list` 等を得て動的に関数呼び出しの定義部分を生成する

```python
import gradio as gr
import openai
import json
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from dotenv import load_dotenv
import os
import asyncio

# 環境変数を読み込み
load_dotenv()

# OpenAI APIキー
openai.api_key = os.getenv("OPENAI_API_KEY")

# MCPサーバーの設定
SERVER_COMMAND = os.getenv("PYTHON_PATH")  # /path/to/venv/bin/python
SERVER_SCRIPT = os.getenv(
    "MCP_SERVER_PATH"
)  # /path/to/mcp_server_youtube/src/mcp_server_youtube/server.py
print(SERVER_COMMAND, SERVER_SCRIPT)

# MCPクライアントのYouTube検索ツール関数

def mcp_search_youtube(query: str, max_results: int = 3):
    """
    MCPサーバー経由でYouTubeを検索（同期版）
    """
    async def async_search():
        server_params = StdioServerParameters(
            command=SERVER_COMMAND, args=[SERVER_SCRIPT], env=None
        )

        # MCPサーバーへの非同期接続
        async with stdio_client(server_params) as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()
                result = await session.call_tool(
                    "youtube-search", arguments={"query": query, "max_results": max_results}
                )
                return result.content[0].text

    # 非同期処理を同期的に実行
    return asyncio.run(async_search())

# Gradioのチャット関数
def chat_with_mcp(history, user_input):
    """
    ChatGPTとの会話を実装（同期版）
    """
    # 初回呼び出し時の履歴初期化
    if history is None:
        history = []

    # 現在の会話履歴を組み立て
    conversation = [
        {
            "role": "system",
            "content": "You are a helpful assistant that can search YouTube videos or have normal conversations.",
        }
    ]
    # Gradioの履歴から会話履歴を変換
    for msg in history:
        conversation.append({"role": msg["role"], "content": msg["content"]})

    # ユーザーの入力を追加
    conversation.append({"role": "user", "content": user_input})

    # ChatGPT APIを呼び出し
    completion = openai.chat.completions.create(
        model="gpt-4o",
        messages=conversation,
        functions=[
            {
                "name": "youtube_search",
                "description": "Search YouTube videos using a query.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "query": {
                            "type": "string",
                            "description": "Search query for YouTube.",
                        },
                        "max_results": {
                            "type": "integer",
                            "description": "Maximum number of results to return.",
                            "default": 3,
                        },
                    },
                    "required": ["query"],
                },
            }
        ],
        function_call="auto",
    )

    # 応答を取得
    message = completion.choices[0].message
    print(message)

    # Function Callingの場合
    if message.function_call:
        print("Function Calling")
        function_call = message.function_call
        function_name = function_call.name
        print(f"Function Name: {function_name}")
        function_args = json.loads(function_call.arguments)

        if function_name == "youtube_search":
            # MCPツールを呼び出し
            query = function_args["query"]
            max_results = function_args.get("max_results", 3)

            # Function呼び出し結果を取得
            search_results = mcp_search_youtube(query, max_results)
            print("Search Results:")
            print(search_results)

            # Functionの結果をconversationに追加
            conversation.append({
                "role": "function",
                "name": function_name,
                "content": search_results,
            })

            # 再度APIを呼び出して、最終的な応答を取得
            completion = openai.chat.completions.create(
                model="gpt-4o",
                messages=conversation,
            )
            assistant_response = completion.choices[0].message.content

            # 履歴を更新
            history.append({"role": "user", "content": user_input})
            history.append({"role": "assistant", "content": assistant_response})
            return history

        else:
            # 未知の関数の場合
            assistant_response = f"Unknown function: {function_name}"
            history.append({"role": "user", "content": user_input})
            history.append({"role": "assistant", "content": assistant_response})
            return history

    else:
        # 通常の応答の場合
        assistant_response = message.content
        history.append({"role": "user", "content": user_input})
        history.append({"role": "assistant", "content": assistant_response})
        return history

# Gradioインターフェース
with gr.Blocks() as demo:
    chatbot = gr.Chatbot(type="messages")
    with gr.Row():
        txt = gr.Textbox(show_label=False, placeholder="Type your message here...")
        submit_btn = gr.Button("Submit")

    submit_btn.click(chat_with_mcp, [chatbot, txt], chatbot)
    txt.submit(chat_with_mcp, [chatbot, txt], chatbot)

# アプリの実行
demo.launch()

```

## まとめ

- MCPはツールをいい感じに定義するやつだぞ
- LLMにサポートしてもらえばわりと一瞬で実装はできるからどんどんやろう