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

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}")

uvx create-mcp-server

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())

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()