Gemini × KlavisでMCPをさくっと使ってみた

GEMINI_API_KEY = "your api key"
KLAVIS_API_KEY = "your api key"

pip install google-genai klavis

import os
from google import genai
from klavis import Klavis
from klavis.types import McpServerName, ToolFormat

YOUTUBE_VIDEO_URL = (
    "https://www.youtube.com/watch?v=_WMjnUpOUi8"  # pick a video you like!
)
SYSTEM_PROMPT = """あなたは、YouTube MCPツールを使用するAIアシスタントです。
利用可能なツールを使って、YouTube動画の情報を取得してください。

動画の要約を求められた場合は：
1. まず、動画のトランスクリプトや詳細情報を取得するツールを使用してください
2. 取得した情報を基に、動画の内容を要約してください

ツールの名前と機能を確認し、適切なツールを選択して使用してください。"""

klavis_api_key = os.environ["KLAVIS_API_KEY"]
klavis_client = Klavis(api_key=klavis_api_key)

google_api_key = os.environ["GEMINI_API_KEY"]
client = genai.Client(api_key=google_api_key)

youtube_mcp_instance = klavis_client.mcp_server.create_server_instance(
    server_name=McpServerName.YOUTUBE,
    user_id="1234",
    platform_name="Klavis",
)


def gemini_with_mcp_server(mcp_server_url: str, user_query: str):
    # Get tools from MCP server
    mcp_server_tools = klavis_client.mcp_server.list_tools(
        server_url=mcp_server_url,
        format=ToolFormat.GEMINI,
    )

    # Send initial message with tools
    response = client.models.generate_content(
        model="gemini-2.0-flash",
        contents=user_query,
        config=genai.types.GenerateContentConfig(
            tools=mcp_server_tools.tools,
            system_instruction=SYSTEM_PROMPT,
        ),
    )

    # Check if function call is requested
    if response.function_calls:
        function_call = response.function_calls[0]
        function_name = function_call.name
        function_args = dict(function_call.args)

        print(f"🔧 Calling: {function_name}, with args: {function_args}")

        # Call the MCP server tool
        result = klavis_client.mcp_server.call_tools(
            server_url=mcp_server_url,
            tool_name=function_name,
            tool_args=function_args,
        )

        function_response = genai.types.Part.from_function_response(
            name=function_call.name,
            response={"result": result.result.content[0]["text"]},
        )
        function_response_content = genai.types.Content(
            role="tool", parts=[function_response]
        )

        # Send function response back to model
        final_response = client.models.generate_content(
            model="gemini-2.0-flash",
            contents=[
                user_query,
                response.candidates[0].content,
                function_response_content,
            ],
            config=genai.types.GenerateContentConfig(
                tools=mcp_server_tools.tools,
            ),
        )
        return final_response.text
    else:
        return response.text


result = gemini_with_mcp_server(
    mcp_server_url=youtube_mcp_instance.server_url,
    user_query=f"このYouTube動画の内容を要約してください: {YOUTUBE_VIDEO_URL}",
)

print(result)

import os
from google import genai
from klavis import Klavis
from klavis.types import McpServerName, ToolFormat
import webbrowser

SYSTEM_PROMPT = """あなたは、Gmail MCPツールを使用するAIアシスタントです。
利用可能なツールを使って、Gmailの情報を取得してください。"""

EMAIL_RECIPIENT = "your_email@gmail.com"  # Replace with your email
EMAIL_SUBJECT = "Test Gemini + Gmail MCP Server"
EMAIL_BODY = "Hello World from Gemini!"

klavis_api_key = os.environ["KLAVIS_API_KEY"]
klavis_client = Klavis(api_key=klavis_api_key)

google_api_key = os.environ["GEMINI_API_KEY"]
client = genai.Client(api_key=google_api_key)

# Create Gmail MCP server instance
gmail_mcp_server = klavis_client.mcp_server.create_server_instance(
    server_name=McpServerName.GMAIL,
    user_id="1234",
    platform_name="Klavis",
)

# Redirect to Gmail OAuth page for authorization
webbrowser.open(gmail_mcp_server.oauth_url)

print(
    f"🔐 Opening OAuth authorization for Gmail, if you are not redirected, please open the following URL in your browser: {gmail_mcp_server.oauth_url}"
)


def gemini_with_mcp_server(mcp_server_url: str, user_query: str):
    # Get tools from MCP server
    mcp_server_tools = klavis_client.mcp_server.list_tools(
        server_url=mcp_server_url,
        format=ToolFormat.GEMINI,
    )

    # Send initial message with tools
    response = client.models.generate_content(
        model="gemini-2.0-flash",
        contents=user_query,
        config=genai.types.GenerateContentConfig(
            tools=mcp_server_tools.tools,
            system_instruction=SYSTEM_PROMPT,
        ),
    )

    # Check if function call is requested
    if response.function_calls:
        function_call = response.function_calls[0]
        function_name = function_call.name
        function_args = dict(function_call.args)

        print(f"🔧 Calling: {function_name}, with args: {function_args}")

        # Call the MCP server tool
        result = klavis_client.mcp_server.call_tools(
            server_url=mcp_server_url,
            tool_name=function_name,
            tool_args=function_args,
        )

        function_response = genai.types.Part.from_function_response(
            name=function_call.name,
            response={"result": result.result.content[0]["text"]},
        )
        function_response_content = genai.types.Content(
            role="tool", parts=[function_response]
        )

        # Send function response back to model
        final_response = client.models.generate_content(
            model="gemini-2.0-flash",
            contents=[
                user_query,
                response.candidates[0].content,
                function_response_content,
            ],
            config=genai.types.GenerateContentConfig(
                tools=mcp_server_tools.tools,
            ),
        )
        return final_response.text
    else:
        return response.text


result = gemini_with_mcp_server(
    mcp_server_url=gmail_mcp_server.server_url,
    user_query=f"Please send an email to {EMAIL_RECIPIENT} with subject {EMAIL_SUBJECT} and body {EMAIL_BODY}",
)

print(result)

webbrowser.open(gmail_mcp_server.oauth_url)