構建 MCP 客戶端-Python

# 創建項目目錄
uv init mcp-client
cd mcp-client

# 創建虛擬環境
uv venv

# 激活虛擬環境
# Windows系統：
.venv\Scripts\activate
# Unix 或 MacOS：
source .venv/bin/activate

# 安裝所需包
uv add mcp anthropic python-dotenv

# 刪除模板文件
rm hello.py

# 創建主文件
touch client.py

# 創建 .env 文件
touch .env

ANTHROPIC_API_KEY=<你的密鑰>

echo ".env" >> .gitignore

import asyncio
from typing import Optional
from contextlib import AsyncExitStack

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

from anthropic import Anthropic
from dotenv import load_dotenv

load_dotenv()  # 從 .env 加載環境變量

class MCPClient:
    def __init__(self):
        # 初始化會話和客戶端對象
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()
        self.anthropic = Anthropic()
    # 方法將在這裡添加

async def connect_to_server(self, server_script_path: str):
    """連接到 MCP 服務器
    
    參數：
        server_script_path: 服務器腳本路徑 (.py 或 .js)
    """
    is_python = server_script_path.endswith('.py')
    is_js = server_script_path.endswith('.js')
    if not (is_python or is_js):
        raise ValueError("服務器腳本必須是 .py 或 .js 文件")
        
    command = "python" if is_python else "node"
    server_params = StdioServerParameters(
        command=command,
        args=[server_script_path],
        env=None
    )
    
    stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
    self.stdio, self.write = stdio_transport
    self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))
    
    await self.session.initialize()
    
    # 列出可用工具
    response = await self.session.list_tools()
    tools = response.tools
    print("\n已連接到服務器，可用工具：", [tool.name for tool in tools])

async def process_query(self, query: str) -> str:
    """使用 Claude 和可用工具處理查詢"""
    messages = [
        {
            "role": "user",
            "content": query
        }
    ]

    response = await self.session.list_tools()
    available_tools = [{ 
        "name": tool.name,
        "description": tool.description,
        "input_schema": tool.inputSchema
    } for tool in response.tools]

    # 初始 Claude API 調用
    response = self.anthropic.messages.create(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1000,
        messages=messages,
        tools=available_tools
    )

    # 處理響應和工具調用
    tool_results = []
    final_text = []

    for content in response.content:
        if content.type == 'text':
            final_text.append(content.text)
        elif content.type == 'tool_use':
            tool_name = content.name
            tool_args = content.input
            
            # 執行工具調用
            result = await self.session.call_tool(tool_name, tool_args)
            tool_results.append({"call": tool_name, "result": result})
            final_text.append(f"[調用工具 {tool_name}，參數 {tool_args}]")

            # 繼續與工具結果的對話
            if hasattr(content, 'text') and content.text:
                messages.append({
                  "role": "assistant",
                  "content": content.text
                })
            messages.append({
                "role": "user", 
                "content": result.content
            })

            # 從 Claude 獲取下一個響應
            response = self.anthropic.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1000,
                messages=messages,
            )

            final_text.append(response.content[0].text)

    return "\n".join(final_text)

async def chat_loop(self):
    """運行交互式聊天循環"""
    print("\nMCP 客戶端已啟動！")
    print("輸入你的問題或輸入 'quit' 退出。")
    
    while True:
        try:
            query = input("\n問題：").strip()
            
            if query.lower() == 'quit':
                break
                
            response = await self.process_query(query)
            print("\n" + response)
                
        except Exception as e:
            print(f"\n錯誤：{str(e)}")

async def cleanup(self):
    """清理資源"""
    await self.exit_stack.aclose()

async def main():
    if len(sys.argv) < 2:
        print("用法：python client.py <服務器腳本路徑>")
        sys.exit(1)
        
    client = MCPClient()
    try:
        await client.connect_to_server(sys.argv[1])
        await client.chat_loop()
    finally:
        await client.cleanup()

if __name__ == "__main__":
    import sys
    asyncio.run(main())

uv run client.py path/to/server.py # python 服務器
uv run client.py path/to/build/index.js # node 服務器

# 相對路徑
uv run client.py ./server/weather.py

# 絕對路徑
uv run client.py /Users/username/projects/mcp-server/weather.py

# Windows 路徑（兩種格式都可以）
uv run client.py C:/projects/mcp-server/weather.py
uv run client.py C:\\projects\\mcp-server\\weather.py