zhangzs/MCP_Client_Server

graph LR
    A[用户] -->|发送请求| B[AI 客户端]
    B -->|查询可用工具| C[MCP 服务器]
    C -->|返回工具列表| B
    B -->|发送请求到 LLM| D[大语言模型]
    D -->|返回工具调用指令| B
    B -->|调用工具| C
    C -->|执行并返回结果| B
    B -->|将结果发送到 LLM| D
    D -->|生成最终回复| B
    B -->|展示结果| A

# Windows (PowerShell)
powershell -c "irm https://astral.sh/uv/install.ps1 | iex"

# macOS/Linux
curl -LsSf https://astral.sh/uv/install.sh | sh

# 初始化项目
uv init mcp-filesystem
cd mcp-filesystem

# 安装依赖
uv add mcp openai python-dotenv

[project]
name = "mcp-client-server"
version = "0.1.0"
description = "MCP 文件系统服务器和客户端示例"
readme = "README.md"
requires-python = ">=3.13"
dependencies = [
    "mcp>=1.17.0",      # MCP 核心库
    "openai>=2.3.0",    # OpenAI SDK（客户端使用）
    "python-dotenv>=1.1.1",  # 环境变量管理
]

from mcp.server.fastmcp import FastMCP
import os

# 创建 MCP 服务器实例，命名为 "filesystem"
mcp = FastMCP("filesystem")

# ==================== 工具定义 ====================

@mcp.tool()
async def read_file(path: str) -> str:
    """
    读取文件内容
    
    参数:
        path: 文件路径（相对或绝对路径）
    返回:
        文件的文本内容
    """
    with open(path, "r", encoding="utf-8") as file:
        return file.read()


@mcp.tool()
async def write_file(path: str, content: str) -> str:
    """
    写入文件内容（覆盖模式）
    
    参数:
        path: 文件路径
        content: 要写入的内容
    返回:
        操作成功提示
    """
    with open(path, "w", encoding="utf-8") as file:
        file.write(content)
    return "File written successfully"


@mcp.tool()
async def list_files(path: str) -> list[str]:
    """
    列出目录中的所有文件和文件夹
    
    参数:
        path: 目录路径
    返回:
        文件和文件夹名称列表
    """
    return os.listdir(path)


@mcp.tool()
async def create_file(path: str, content: str) -> str:
    """
    创建新文件并写入内容
    
    参数:
        path: 新文件路径
        content: 初始内容
    返回:
        操作成功提示
    """
    with open(path, "w", encoding="utf-8") as file:
        file.write(content)
    return "File created successfully"


@mcp.tool()
async def delete_file(path: str) -> str:
    """
    删除文件
    
    参数:
        path: 要删除的文件路径
    返回:
        操作成功提示
    """
    os.remove(path)
    return "File deleted successfully"


@mcp.tool()
async def rename_file(path: str, new_path: str) -> str:
    """
    重命名或移动文件
    
    参数:
        path: 原文件路径
        new_path: 新文件路径
    返回:
        操作成功提示
    """
    os.rename(path, new_path)
    return "File renamed successfully"


# ==================== 资源定义 ====================

@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
    """
    获取个性化问候语（Resource 示例）
    
    参数:
        name: 用户名称
    返回:
        问候语文本
    """
    return f"Hello, {name}!"


# ==================== 启动服务器 ====================

if __name__ == "__main__":
    print("MCP server is running...")
    mcp.run(transport="stdio")  # 使用标准输入输出通信

npm install -g @modelcontextprotocol/inspector

mcp-inspector uv run mcp_server.py

{
  "path": "."
}

{
  "path": "test.txt",
  "content": "Hello from MCP!"
}

ASE_URL=https://api.deepseek.com
MODEL=deepseek-chat
API_KEY="sk-xxxxxxxxxx"

import os
import asyncio
from openai import OpenAI
from dotenv import load_dotenv
from contextlib import AsyncExitStack
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import json

# 加载环境变量
load_dotenv()

# 获取配置
base_url = os.getenv("BASE_URL")
model = os.getenv("MODEL")
api_key = os.getenv("API_KEY")


class MCPClient:
    """MCP 客户端：连接 MCP 服务器和大语言模型"""
    
    def __init__(self):
        """初始化客户端"""
        self.exit_stack = AsyncExitStack()  # 用于管理异步资源
        self.openai_api_key = api_key
        self.base_url = base_url
        self.model = model
        
        # 验证环境变量
        if not all([self.openai_api_key, self.base_url, self.model]):
            raise ValueError("Missing required environment variables")
        
        # 创建 OpenAI 客户端
        self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url)
    
    
    async def __aenter__(self):
        """进入异步上下文管理器"""
        return self
    
    
    async def __aexit__(self, exc_type, exc_value, traceback):
        """退出上下文时清理所有资源"""
        # 1. 关闭 OpenAI 客户端
        try:
            if hasattr(self, 'client') and self.client is not None:
                self.client.close()
                print("[OK] OpenAI 客户端已关闭")
        except Exception as e:
            print(f"[WARNING] 关闭 OpenAI 客户端时出错: {e}")
        
        # 2. 清理 MCP 连接
        try:
            await self.exit_stack.aclose()
            print("[OK] MCP 连接已关闭")
        except Exception as e:
            print(f"[WARNING] 关闭 MCP 连接时出错: {e}")
    
    
    async def connect_to_server(self, server_url: str):
        """
        连接到 MCP 服务器
        
        参数:
            server_url: 服务器脚本路径（.py 或 .js 文件）
        """
        # 判断服务器类型
        is_python = server_url.endswith(".py")
        
        # 转换为绝对路径
        if not os.path.isabs(server_url):
            server_url = os.path.abspath(server_url)
        
        # 配置启动命令
        if is_python:
            command = "uv"
            args = ["run", server_url]
        else:
            command = "node"
            args = [server_url]
        
        # 创建服务器参数
        server_params = StdioServerParameters(
            command=command, 
            args=args, 
            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()
    
    
    async def list_tools(self):
        """列出服务器提供的所有工具"""
        response = await self.session.list_tools()
        tools = response.tools
        print("已连接服务器，支持的工具如下：", [tool.name for tool in tools])
    
    
    async def process_query(self, query: str) -> str:
        """
        处理用户查询：调用 LLM 并执行工具
        
        参数:
            query: 用户的自然语言请求
        返回:
            AI 助手的回复
        """
        # 构建对话消息
        messages = [
            {
                "role": "system", 
                "content": """你是一个智能助手，可以帮助用户操作文件系统。

重要规则：
1. 当用户要求写入特定内容时，必须完全按照用户提供的原文逐字写入
2. 请仔细检查并保留用户输入的原文，确保每个字都准确无误
3. 特别注意中文内容，不要改变、替换或简化任何字词"""
            },
            {"role": "user", "content": query}
        ]
        
        try:
            # 1. 获取 MCP 服务器的工具列表
            response = await self.session.list_tools()
            available_tools = [{
                'type': 'function',
                'function': {
                    'name': tool.name,                # 工具名称
                    'description': tool.description,  # 工具描述
                    'parameters': tool.inputSchema    # 参数定义（JSON Schema）
                }
            } for tool in response.tools]
            
            # 2. 调用 LLM，传递工具信息
            # 使用 run_in_executor 在线程池中运行同步的 OpenAI 调用
            response = await asyncio.get_event_loop().run_in_executor(
                None,
                lambda: self.client.chat.completions.create(
                    model=self.model,
                    messages=messages,
                    tools=available_tools  # 告诉 LLM 可用的工具
                )
            )
            
            content = response.choices[0]
            
            # 3. 检查 LLM 是否要求调用工具
            if content.finish_reason == "tool_calls":
                # 提取工具调用信息
                tool_call = content.message.tool_calls[0]
                tool_name = tool_call.function.name
                tool_args = json.loads(tool_call.function.arguments)
                print(f"调用工具：{tool_name}, 参数：{tool_args}")
                
                # 4. 执行 MCP 工具调用
                result = await self.session.call_tool(tool_name, tool_args)
                
                # 5. 将工具调用和结果添加到对话历史
                messages.append(content.message.model_dump())
                messages.append({
                    "role": "tool",
                    "content": result.content[0].text,
                    "tool_call_id": tool_call.id
                })
                
                # 6. 再次请求 LLM，让它基于工具结果生成最终回复
                response = await asyncio.get_event_loop().run_in_executor(
                    None,
                    lambda: self.client.chat.completions.create(
                        model=self.model,
                        messages=messages,
                        tools=available_tools
                    )
                )
            
            # 7. 返回 AI 的最终回复
            return response.choices[0].message.content
        
        except Exception as e:
            print(f"Error processing query: {e}")
            return f"Error: {e}"
    
    
    async def chat_loop(self):
        """聊天循环：持续接收用户输入"""
        print("\n欢迎使用智能助手，输入 'exit' 退出\n")
        
        while True:
            try:
                query = input("You: ")
                if query.lower() == "exit":
                    print("再见！")
                    break
                
                response = await self.process_query(query)
                print(f"Assistant: {response}\n")
            
            except Exception as e:
                print(f"Error: {e}\n")


async def main():
    """主函数：启动客户端"""
    async with MCPClient() as client:
        # 连接到 MCP 服务器
        await client.connect_to_server("mcp_server.py")
        
        # 列出可用工具
        await client.list_tools()
        
        # 进入聊天循环
        await client.chat_loop()


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

# 核心步骤：
# 1. 配置服务器启动命令（uv run for Python, node for JS）
# 2. 创建 StdioServerParameters（标准输入输出通信参数）
# 3. 使用 stdio_client 启动服务器进程
# 4. 创建 ClientSession 建立会话
# 5. 调用 initialize() 完成握手

# 检查 LLM 是否要求调用工具
if content.finish_reason == "tool_calls":
    # 提取工具名称和参数
    tool_call = content.message.tool_calls[0]
    tool_name = tool_call.function.name
    tool_args = json.loads(tool_call.function.arguments)
    
    # 调用 MCP 工具
    result = await self.session.call_tool(tool_name, tool_args)
    
    # 构建工具结果消息
    messages.append({
        "role": "tool",
        "content": result.content[0].text,
        "tool_call_id": tool_call.id
    })

response = await asyncio.get_event_loop().run_in_executor(
    None,  # None 表示使用默认的 ThreadPoolExecutor
    lambda: self.client.chat.completions.create(...)
)

# 进入上下文时资源会被注册
stdio_transport = await self.exit_stack.enter_async_context(
    stdio_client(server_params)
)

# 退出上下文时自动清理所有资源
async def __aexit__(self, ...):
    await self.exit_stack.aclose()  # 按相反顺序关闭所有资源

python llm_client.py

# 1. 路径验证：防止目录遍历攻击
import os
def is_safe_path(base_dir, path):
    """检查路径是否在允许的目录内"""
    abs_path = os.path.abspath(path)
    abs_base = os.path.abspath(base_dir)
    return abs_path.startswith(abs_base)

# 2. 权限控制：限制可访问的目录
ALLOWED_DIRS = ["/workspace", "/data"]

# 3. 参数验证：使用 Pydantic 模型
from pydantic import BaseModel, Field

class ReadFileArgs(BaseModel):
    path: str = Field(..., description="文件路径", pattern=r"^[a-zA-Z0-9_./\-]+$")

@mcp.tool()
async def read_file(path: str) -> str:
    try:
        if not os.path.exists(path):
            return f"错误：文件 {path} 不存在"
        
        with open(path, "r", encoding="utf-8") as file:
            return file.read()
    
    except PermissionError:
        return f"错误：没有权限读取 {path}"
    except Exception as e:
        return f"读取文件时出错：{str(e)}"

# 1. 对于大文件，使用流式读取
@mcp.tool()
async def read_large_file(path: str, max_size: int = 1024 * 1024) -> str:
    """读取文件（最多 1MB）"""
    with open(path, "r", encoding="utf-8") as file:
        return file.read(max_size)

# 2. 使用异步 I/O（aiofiles）
import aiofiles

@mcp.tool()
async def read_file_async(path: str) -> str:
    async with aiofiles.open(path, "r", encoding="utf-8") as file:
        return await file.read()