客户端开发

# 创建项目目录
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 main.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模型和MCP服务器上的可用工具处理用户查询"""
    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
    )
 
    # 处理Claude响应和可能的工具调用
    final_text = []
 
    assistant_message_content = []
    for content in response.content:
        if content.type == 'text':
            # 处理纯文本回复
            final_text.append(content.text)
            assistant_message_content.append(content)
        elif content.type == 'tool_use':
            # 处理工具调用请求
            tool_name = content.name
            tool_args = content.input
 
            # 通过MCP服务器执行工具调用
            result = await self.session.call_tool(tool_name, tool_args)
            final_text.append(f"[调用工具: {tool_name}，参数: {tool_args}]")
 
            # 更新对话历史
            assistant_message_content.append(content)
            messages.append({
                "role": "assistant",
                "content": assistant_message_content
            })
            messages.append({
                "role": "user",
                "content": [
                    {
                        "type": "tool_result",
                        "tool_use_id": content.id,
                        "content": result.content
                    }
                ]
            })
 
            # 将工具执行结果发送回Claude，获取进一步回复
            response = self.anthropic.messages.create(
                model="claude-3-5-sonnet-20241022",
                max_tokens=1000,
                messages=messages,
                tools=available_tools
            )
 
            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