"""
MyMCP 用戶端 - 使用 OpenAI 原生 tools 調用
https://www.claudemcp.com/zh/docs/mcp-py-sdk-basic
"""
import asyncio
import json
import os
import sys
from typing import Dict, List, Any, Optional
from dataclasses import dataclass
from openai import AsyncOpenAI
from mcp import StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.client.session import ClientSession
from mcp.types import Tool, TextContent
from rich.console import Console
from rich.prompt import Prompt
from rich.panel import Panel
from rich.markdown import Markdown
from rich.table import Table
from rich.spinner import Spinner
from rich.live import Live
from dotenv import load_dotenv
# 載入環境變數
load_dotenv()
# 初始化 Rich console
console = Console()
@dataclass
class MCPServerConfig:
"""MCP 伺服器設定"""
name: str
command: str
args: List[str]
description: str
env: Optional[Dict[str, str]] = None
class MyMCPClient:
"""MyMCP 用戶端"""
def __init__(self, config_path: str = "mcp.json"):
self.config_path = config_path
self.servers: Dict[str, MCPServerConfig] = {}
self.all_tools: List[tuple[str, Any]] = [] # (server_name, tool)
self.openai_client = AsyncOpenAI(
api_key=os.getenv("OPENAI_API_KEY")
)
def load_config(self):
"""從設定檔載入 MCP 伺服器設定"""
try:
with open(self.config_path, 'r', encoding='utf-8') as f:
config = json.load(f)
for name, server_config in config.get("mcpServers", {}).items():
env_dict = server_config.get("env", {})
self.servers[name] = MCPServerConfig(
name=name,
command=server_config["command"],
args=server_config.get("args", []),
description=server_config.get("description", ""),
env=env_dict if env_dict else None
)
console.print(f"[green]✓ 已載入 {len(self.servers)} 個 MCP 伺服器設定[/green]")
except Exception as e:
console.print(f"[red]✗ 載入設定檔失敗: {e}[/red]")
sys.exit(1)
async def get_tools_from_server(self, name: str, config: MCPServerConfig) -> List[Tool]:
"""從單一伺服器取得工具清單"""
try:
console.print(f"[blue]→ 正在連接伺服器: {name}[/blue]")
# 準備環境變數
env = os.environ.copy()
if config.env:
env.update(config.env)
# 建立伺服器參數
server_params = StdioServerParameters(
command=config.command,
args=config.args,
env=env
)
# 使用 async with 上下文管理器（雙層嵌套）
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# 取得工具列表
tools_result = await session.list_tools()
tools = tools_result.tools
console.print(f"[green]✓ {name}: {len(tools)} 個工具[/green]")
return tools
except Exception as e:
console.print(f"[red]✗ 連線伺服器 {name} 失敗: {e}[/red]")
console.print(f"[red] 錯誤類型: {type(e).__name__}[/red]")
import traceback
console.print(f"[red] 詳細錯誤: {traceback.format_exc()}[/red]")
return []
async def load_all_tools(self):
"""載入所有伺服器的工具"""
console.print("\n[blue]→ 正在取得可用工具清單...[/blue]")
for name, config in self.servers.items():
tools = await self.get_tools_from_server(name, config)
for tool in tools:
self.all_tools.append((name, tool))
def display_tools(self):
"""顯示所有可用工具"""
table = Table(title="可用 MCP 工具", show_header=True)
table.add_column("伺服器", style="cyan")
table.add_column("工具名稱", style="green")
table.add_column("描述", style="white")
# 按伺服器分組
current_server = None
for server_name, tool in self.all_tools:
# 只在伺服器名稱變更時顯示伺服器名稱
display_server = server_name if server_name != current_server else ""
current_server = server_name
table.add_row(
display_server,
tool.name,
tool.description or "無描述"
)
console.print(table)
def build_openai_tools(self) -> List[Dict[str, Any]]:
"""建構 OpenAI tools 格式的工具定義"""
openai_tools = []
for server_name, tool in self.all_tools:
# 建構 OpenAI function 格式
function_def = {
"type": "function",
"function": {
"name": f"{server_name}_{tool.name}", # 新增伺服器前綴避免衝突
"description": f"[{server_name}] {tool.description or '無描述'}",
"parameters": tool.inputSchema or {"type": "object", "properties": {}}
}
}
openai_tools.append(function_def)
return openai_tools
def parse_tool_name(self, function_name: str) -> tuple[str, str]:
"""解析工具名稱，提取伺服器名稱和工具名稱"""
# 格式: server_name_tool_name
parts = function_name.split('_', 1)
if len(parts) == 2:
return parts[0], parts[1]
else:
# 如果沒有底線，假設是第一個伺服器的工具
if self.all_tools:
return self.all_tools[0][0], function_name
return "unknown", function_name
async def call_tool(self, server_name: str, tool_name: str, arguments: Dict[str, Any]) -> Any:
"""呼叫指定的工具"""
config = self.servers.get(server_name)
if not config:
raise ValueError(f"伺服器 {server_name} 不存在")
try:
# 準備環境變數
env = os.environ.copy()
if config.env:
env.update(config.env)
# 建立伺服器參數
server_params = StdioServerParameters(
command=config.command,
args=config.args,
env=env
)
# 使用 async with 上下文管理器（雙層嵌套）
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize()
# 調用工具
result = await session.call_tool(tool_name, arguments)
return result
except Exception as e:
console.print(f"[red]✗ 呼叫工具 {tool_name} 失敗: {e}[/red]")
raise
def extract_text_content(self, content_list: List[Any]) -> str:
"""從 MCP 回應中提取文字內容"""
text_parts: List[str] = []
for content in content_list:
if isinstance(content, TextContent):
text_parts.append(content.text)
elif hasattr(content, 'text'):
text_parts.append(str(content.text))
else:
# 處理其他類型的內容
text_parts.append(str(content))
return "\n".join(text_parts) if text_parts else "✅ 操作完成，但沒有回傳文字內容"
async def process_user_input(self, user_input: str) -> str:
"""處理使用者輸入並返回最終回應"""
# 建置工具定義
openai_tools = self.build_openai_tools()
try:
# 第一次呼叫 - 讓 LLM 決定是否需要使用工具
messages = [
{"role": "system", "content": "你是個智慧助手，可以使用各種 MCP 工具來幫助使用者完成任務。如果不需要使用工具，直接返回回答。"},
{"role": "user", "content": user_input}
]
# 呼叫 OpenAI API
kwargs = {
"model": "gpt-4.1",
"messages": messages,
"temperature": 0.7
}
# 只有當有工具時才加入 tools 參數
if openai_tools:
kwargs["tools"] = openai_tools
kwargs["tool_choice"] = "auto"
# 使用 loading 特效
with Live(Spinner("dots", text="[blue]正在思考...[/blue]"), console=console, refresh_per_second=10):
response = await self.openai_client.chat.completions.create(**kwargs) # type: ignore
message = response.choices[0].message
# 檢查是否有工具調用
if hasattr(message, 'tool_calls') and message.tool_calls: # type: ignore
# 新增助手訊息到歷史
messages.append({ # type: ignore
"role": "assistant",
"content": message.content,
"tool_calls": [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments
}
} for tc in message.tool_calls # type: ignore
]
})
# 執行每個工具調用
for tool_call in message.tool_calls:
function_name = tool_call.function.name # type: ignore
arguments = json.loads(tool_call.function.arguments) # type: ignore
# 解析伺服器名稱和工具名稱
server_name, tool_name = self.parse_tool_name(function_name) # type: ignore
try:
# 使用 loading 特效呼叫工具
with Live(Spinner("dots", text=f"[cyan]正在呼叫 {server_name}.{tool_name}...[/cyan]"), console=console, refresh_per_second=10):
result = await self.call_tool(server_name, tool_name, arguments)
# 從 MCP 回應中提取文字內容
result_content = self.extract_text_content(result.content)
# 新增工具呼叫結果
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": result_content
})
console.print(f"[green]✓ {server_name}.{tool_name} 呼叫成功[/green]")
except Exception as e:
# 新增錯誤訊息
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": f"錯誤: {str(e)}"
})
console.print(f"[red]✗ {server_name}.{tool_name} 呼叫失敗: {e}[/red]")
# 取得最終回應
with Live(Spinner("dots", text="[blue]正在產生最終回應...[/blue]"), console=console, refresh_per_second=10):
final_response = await self.openai_client.chat.completions.create(
model="gpt-4.1",
messages=messages, # type: ignore
temperature=0.7
)
final_content = final_response.choices[0].message.content
return final_content or "抱歉，我無法產生最終答案。"
else:
# 沒有工具調用，直接回傳回應
return message.content or "抱歉，我無法產生答案。"
except Exception as e:
console.print(f"[red]✗ 處理請求時發生錯誤: {e}[/red]")
return f"抱歉，處理您的請求時出現錯誤: {str(e)}"
async def interactive_loop(self):
"""互動式迴圈"""
console.print(Panel.fit(
"[bold cyan]MyMCP 用戶端已啟動[/bold cyan]\n"
"輸入您的問題，我會使用可用的 MCP 工具來幫助您。\n"
"輸入 'tools' 查看可用工具\n"
"輸入 'exit' 或 'quit' 退出。",
title="歡迎使用 MCP 用戶端"
))
while True:
try:
# 取得用戶輸入
user_input = Prompt.ask("\n[bold green]您[/bold green]")
if user_input.lower() in ['exit', 'quit', 'q']:
console.print("\n[yellow]再見！[/yellow]")
break
if user_input.lower() == 'tools':
self.display_tools()
continue
# 處理使用者輸入
response = await self.process_user_input(user_input)
# 顯示回應
console.print("\n[bold blue]助手[/bold blue]:")
console.print(Panel(Markdown(response), border_style="blue"))
except KeyboardInterrupt:
console.print("\n[yellow]已中斷[/yellow]")
break
except Exception as e:
console.print(f"\n[red]錯誤: {e}[/red]")
async def run(self):
"""運行客戶端"""
# 載入配置
self.load_config()
if not self.servers:
console.print("[red]✗ 沒有設定的伺服器[/red]")
return
# 取得所有工具
await self.load_all_tools()
if not self.all_tools:
console.print("[red]✗ 沒有可用的工具[/red]")
return
# 顯示可用工具
self.display_tools()
# 進入交互循環
await self.interactive_loop()
async def main():
"""主函式"""
# 檢查 OpenAI API Key
if not os.getenv("OPENAI_API_KEY"):
console.print("[red]✗ 請設定環境變數 OPENAI_API_KEY[/red]")
console.print("提示: 建立 .env 檔案並新增: OPENAI_API_KEY=your-api-key")
sys.exit(1)
# 創建並運行客戶端
client = MyMCPClient()
await client.run()
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
console.print("\n[yellow]程式已退出[/yellow]")
except Exception as e:
console.print(f"\n[red]程式錯誤: {e}[/red]")
sys.exit(1)
