# #!/usr/bin/env python3
# """
# Gemini APIを使用したMCPクライアント
# 自然言語でユーザーの質問を処理し、必要に応じてMCPサーバーの
# ツールを呼び出すインテリジェントクライアント
# """
import asyncio
import json
import os
from pathlib import Path
from typing import Any, Dict, List
from dotenv import load_dotenv
from fastmcp import Client
from google import genai
class MCPClient:
"""Simple MCP HTTP Client"""
def __init__(self, base_url: str):
self.base_url = base_url
self.timeout = 9
async def list_tools(self) -> List[Dict[str, Any]]:
"""利用可能なツールの一覧を取得"""
async with Client(self.base_url) as client:
return await client.list_tools()
async def call_tool(self, name: str, arguments: Dict[str, Any]) -> Any:
"""ツールを呼び出し"""
async with Client(self.base_url) as client:
return await client.call_tool(name, arguments)
class GeminiMCPClient:
"""Gemini APIを使用したインテリジェントMCPクライアント"""
def __init__(self, mcp_client: MCPClient, gemini_api_key: str):
self.mcp_client = mcp_client
self.llm_client = genai.Client(api_key=gemini_api_key)
self.tools_info = None
async def initialize(self):
"""ツール情報を取得して初期化"""
self.tools_info = await self.mcp_client.list_tools()
def create_system_prompt(self) -> str:
"""システムプロンプトを作成"""
tools_description = ""
if self.tools_info:
for tool in self.tools_info:
name = tool.name
description = tool.description
tools_description += f"- {name}: {description}\n"
return f"""あなたは日本の天気情報を提供するアシスタントです。
利用可能なツール:
{tools_description}
ユーザーの質問を分析し、適切なツールを呼び出してください。
以下のJSON形式でツール呼び出しの指示を出力してください:
```json
{{
"action": "call_tool",
"tool_name": "ツール名",
"arguments": {{
"location": "地域名",
"days": 日数 (オプション)
}}
}}
```
または、ツール呼び出しが不要な場合:
```json
{{
"action": "respond",
"message": "直接回答"
}}
```
日本語で自然に回答してください。"""
async def process_query(self, user_query: str) -> str:
"""ユーザーの質問を処理"""
system_prompt = self.create_system_prompt()
full_prompt = f"""{system_prompt}
ユーザーの質問: {user_query}
上記の質問に対して、必要であればツールを呼び出し、適切に回答してください。"""
try:
# Gemini APIで質問を解析
response = self.llm_client.models.generate_content(
model="gemini-1.5-flash", contents=full_prompt
)
gemini_response = response.text.strip()
# JSON部分を抽出
json_start = gemini_response.find("```json")
json_end = gemini_response.find("```", json_start + 7)
if json_start != -1 and json_end != -1:
json_content = gemini_response[json_start + 7 : json_end].strip()
try:
action_data = json.loads(json_content)
if action_data.get("action") == "call_tool":
# ツールを呼び出し
tool_name = action_data.get("tool_name")
arguments = action_data.get("arguments", {})
print(f"🔧 ツール呼び出し: {tool_name} with {arguments}")
tool_result = await self.mcp_client.call_tool(
tool_name, arguments
)
# 結果を元に自然な回答を生成
result_prompt = f"""以下のツール実行結果を元に、ユーザーに分かりやすく日本語で回答してください。
ユーザーの質問: {user_query}
ツール実行結果: {json.dumps(tool_result.data, ensure_ascii=False, indent=2)}
自然で親しみやすい回答をしてください。"""
final_response = self.llm_client.models.generate_content(
model="gemini-1.5-flash", contents=result_prompt
)
return final_response.text.strip()
elif action_data.get("action") == "respond":
return action_data.get(
"message", "申し訳ありませんが、回答できませんでした。"
)
except json.JSONDecodeError:
pass
# JSONが解析できない場合はそのまま返す
return gemini_response
except Exception as e:
return f"申し訳ありませんが、エラーが発生しました: {str(e)}"
async def load_env_config():
"""環境設定を読み込み"""
# .envファイルを探して読み込み
current_dir = Path(__file__).parent
env_file = current_dir / ".env"
if env_file.exists():
load_dotenv(env_file)
print(f"✅ 設定ファイル読み込み: {env_file}")
else:
print(f"⚠️ .envファイルが見つかりません: {env_file}")
print("環境変数または.envファイルでAPIキーを設定してください")
async def interactive_gemini_client():
"""Gemini APIを使用したインタラクティブクライアント"""
# 環境設定の読み込み
await load_env_config()
# API設定の確認
gemini_api_key = os.getenv("GEMINI_API_KEY")
if not gemini_api_key:
print("❌ GEMINI_API_KEY環境変数が設定されていません。")
print("以下のいずれかの方法で設定してください:")
print("1. examples/.envファイルを作成してAPIキーを設定")
print("2. 環境変数で設定: export GEMINI_API_KEY=your_api_key_here")
print("APIキーは Google AI Studio で取得できます:")
print("https://makersuite.google.com/app/apikey")
return
server_url = os.getenv("MCP_SERVER_URL", "http://127.0.1:8000/mcp/")
print(
"🤖 Gemini + Japanese Weather MCP Client"
"を起動しています..."
f" (server: {server_url})"
)
try:
# MCPクライアントとGeminiクライアントを初期化
mcp_client = MCPClient(server_url)
gemini_client = GeminiMCPClient(mcp_client, gemini_api_key)
print("🔄 MCPサーバーに接続中...")
await gemini_client.initialize()
print("✅ 準備完了！")
while True:
try:
user_input = input("\n💬 質問: ").strip()
if user_input.lower() in ["quit", "exit", "終了"]:
print("👋 ありがとうございました！")
break
if not user_input:
continue
print("🤔 考え中...")
response = await gemini_client.process_query(user_input)
print(f"\n🤖 回答:\n{response}")
except KeyboardInterrupt:
print("\n👋 ありがとうございました！")
break
except Exception as e:
print(f"❌ エラー: {e}")
except Exception as e:
print(f"❌ 初期化エラー: {e}")
print("MCPサーバーが起動していることを確認してください:")
async def main():
"""メイン関数"""
await interactive_gemini_client()
if __name__ == "__main__":
asyncio.run(main())
