MPC

MCP概述

MCP运行机制

MCP 整体架构

MCP 核心概念

1. MCP 主机(MCP Hosts): 作为运行 MCP 的主应用程序，例如 Claude Desktop、Cursor、Cline 或 AI 工具。 为用户提供与LLM交互的接口，同时集成 MCP Client 以连接 MCP Server。

2. MCP 客户端( MCP Clients ): MCP client 充当 LLM 和 MCP server 之间的桥梁，嵌入在主机程序中。
• 主要职责：
• 接收来自LLM的请求；
• 将请求转发到相应的 MCP server
• 将 MCP server 的结果返回给 LLM
• 客户端类型：
• 聊天客户端：Cherry Studio、Claude、Librechat、Chatwise等
• AI编程IDE：Cursor、Cline、Continue、Sourcegraph、Windsurf 等

3. MCP 服务器( MCP Servers ): 每个 MCP 服务器都提供了一组特定的工具，负责从本地数据或远程服务中检索信息。 是 MCP 架构中的关键组件。与传统的远程 API 服务器不同，MCP 服务器既可以作为本地应用程序在用户设备上运行（与客户端以stdio的方式交互），也可部署至远程服务器（与客户端以SSE的方式交互）。MCP 服务器的本质是运行在电脑上的一个nodejs或python程序。可以理解为客户端用命令行调用了 电脑上的nodejs或python程序。
• 使用 TypeScript 编写的 MCP server 可以通过 npx 命令来运行
• 使用 Python 编写的 MCP server 可以通过 uvx 命令来运行。

4. 本地资源( Local Resources ): Your computer’s files, databases, and services that MCP servers can securely access；本地数据源：您的计算机文件、数据库以及 MCP 服务器可以安全访问的服务

5. 远程服务( Remote Resources ): External systems available over the internet (e.g., through APIs) that MCP servers can connect to；远程服务：可通过互联网（例如，通过 API）访问的外部系统，MCP 服务器可以连接到

MCP 通信方式

• stdio（Standard Input/Output）：传输允许通过标准输入和输出流进行通信。主要用在本地服务上，操作本地的软件或者本地的文件。这对于本地集成和命令行工具特别有用。例如：
• 构建命令行工具
• 实现本地集成
• 需要简单的进程通信
• 与 shell 脚本一起工作

• HTTP（Streamable HTTP）：可流式 HTTP 传输正在取代 SSE 传输用于生产部署
• 有状态和无状态操作模式
• 可恢复性与事件存储
• JSON 或 SSE 响应格式
• 更好的多节点部署可扩展性

在IDE中配置 MCP

环境准备

• Python 中需要安装 uv ；注意：需要安装到全局环境中，如果只安装到MCP 服务项目的虚拟环境中不行；Mac：brew install uv
• 安装好之后对应的执行指令为：uvx

• Typescript 中需要安装 Node.js；
• 安装好之后对应的执行指令为：npx

VSCode中的cline插件

• Cline 官方文档：‣

• Cline 官方配置教程：‣

1. 在 VSCode 中安装 Cline 插件

2. 配置自己的模型 API

3. 进入 MCP 配置文件

4. 编辑配置文件
• 进入之后是一个 json 文件，格式如下：
本地服务格式
{
  "mcpServers": {
    "your-mcp-identifier": {
      "command": "your-command",
      "args": ["arg1", "arg2"],
      "env": {
        "YOUR_API_KEY_NAME": "your_api_key"
      },
      "autoApprove": ["tool1", "tool2"],
      "disabled": false
    }
  }
}
格式说明：
远程服务格式

{
  "mcpServers": {
    "remote-server": {
      "url": "https://your-server-url.com/mcp",
      "headers": {
        "Authorization": "Bearer your-token"
      },
      "autoApprove": ["tool3"],
      "disabled": false
    }
  }
}

• 本地 MCP 服务实例

{
  "mcpServers": {
    "bilibili_server": {
      "command": "/Users/ayd/miniconda3/envs/mcp/bin/python",
      "args": ["/Users/ayd/Desktop/MCP/res/mcp_bilibili_server.py"]
    }
  }
}

注意：因为开发的是 Python 脚本，所以 "command" 要写对应环境变量中的 python 路径，"args" 中的第一个参数应该是对应的脚本

5. 状态检查

6. 开始智能体问答（在 act 模式下）
中途会询问是否允许关键词搜索，点击允许，即可得到对应的内容回答

Cursor

全局配置（不推荐）

项目配置

在代码中开发MCP

常用的 MCP 服务器

• 文件系统 filesystem
• Filesystem MCP 旨在为大型语言模型(LLM)和 AI 助手提供对本地文件系统的安全、受控访问。 主要功能： - 文件读写：允许读取和写入文件内容，支持创建新文件或覆盖现有文件。 - 目录管理：支持创建、列出和删除目录，以及移动文件或目录。 - 文件搜索：能够在指定路径中搜索匹配特定模式的文件或目录。 - 元数据获取：提供获取文件或目录的详细元数据，包括大小、创建时间、修改时间、访问时间、类型和权限等信息。

• 数据库 mysqldb-mcp-server
• 一种模型上下文协议 （MCP） 实现，支持与 MySQL 数据库进行安全交互。此服务器组件可促进 AI 应用程序（主机/客户端）与 MySQL 数据库之间的通信，提供安全的 MySQL 数据库操作，通过受控接口使数据库探索和分析更安全、更有条理。

• 网页数据采集 Firecrawl
• Firecrawl MCP 工具是一款基于模型上下文协议（MCP）的企业级网页数据采集服务器。能够为大型语言模型（LLM）提供强大的网页抓取能力。 主要功能： • JavaScript 渲染：能够处理动态网页内容，突破传统抓取工具的局限，获取更全面的数据。 • 批量处理：支持并行处理和队列管理，提高数据抓取效率。 • 智能限速：根据网络状况和任务需求智能调整抓取速度，避免对目标网站造成过大压力。 • 多种输出格式：支持将抓取的内容转换为 Markdown、HTML 等格式，满足不同场景的需求。 说明：去firecrawl官网注册后即可查看自己的api_key

• Github
• GitHub MCP 服务器是一个模型上下文协议 （MCP） 提供与 GitHub API 无缝集成的服务器，从而实现面向开发人员的高级自动化工具和交互功能。 使用案例： • 自动化 GitHub 工作流程和流程。 • 从 GitHub 存储库中提取和分析数据。 • 构建与 GitHub 生态系统交互的 AI 驱动的工具和应用程序。 说明：去https://github.com/settings/tokens 申请自己的token

• 控制台 desktop-commander
• 在计算机上无缝执行终端命令和管理流程。使用强大的命令执行和文件作工具简化您的开发任务。

MCP 服务端

学习资源

• ‣

• ‣

要点总结

• 在写 mcp.tool 的时候要添加详细的函数摘要，这个摘要最终会提供给客户端的大模型，让它来结合用户的输入来判断到底要使用哪个 tool

• 开发完成之后使用 mcp dev xxx.py 命令来气筒服务测试

• MCP 服务器可以提供三种主要类型的能力，分别对应不同的装饰器
• Resources：资源：客户端可读取的数据（如 API 响应或文件内容）；mcp.resource
• Tools：工具：可由LLM（经用户同意）调用的函数；mcp.tool
• Prompts：提示词：帮助用户完成特定任务的预写模板；mcp.prompt

代码示例

import re
import httpx
from typing import Any
from mcp.server.fastmcp import FastMCP


mcp = FastMCP('weather')
# mcp = FastMCP("weather", stateless_http=True)
# stateless_http：无状态；每次请求都会创建一个新的临时连接，请求之间不维护会话状态，任务生命周期局限于单个请求，适用于多节点环境部署

OPENWEATHER_API_BASE = 'https://api.openweathermap.org/data/2.5'
OPENWEATHER_API_KEY = '58c8d33967e43de50488daaddc629bc1'
USER_AGENT = "weather-app/1.0"

async def make_weather_request(url: str) -> dict[str, Any] | None:
    """向OpenWeatherMap API 请求天气数据."""
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/json",
        "Content-Type": "application/json",
    }
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, headers=headers, timeout=30.0)
            response.raise_for_status()
            return response.json()
        except Exception:
            return None

def format_weather_data(data: dict, units: str = "metric") -> str:
    """将JSON格式的天气数据转换为可读的文本字符串格式"""
    # 检查data是否为空
    if not data:
        return "无法获取天气数据"
    
    # 从 API 响应中提取关键天气信息
    country_code = data.get("sys", {}).get("country", "未知")               # 获取国家代码
    city_name = data.get("name", "未知")                                    # 获取城市名称
    weather_desc = data.get("weather", [{}])[0].get("description", "未知")  # 获取天气描述
    temp = data.get("main", {}).get("temp", "未知")                         # 获取当前温度（摄氏度）
    feels_like = data.get("main", {}).get("feels_like", "未知")             # 获取体感温度（摄氏度）
    humidity = data.get("main", {}).get("humidity", "未知")                 # 获取湿度(百分比)
    wind_speed = data.get("wind", {}).get("speed", "未知")                  # 获取风速(m/s)

    # 根据 units 参数确定温度和风速的单位
    temp_unit = "°C" if units == "metric" else "°F"
    wind_speed_unit = "m/s" if units == "metric" else "mph"

    # 构建格式化的天气信息字符串
    return f"""
    城市: {city_name}
    国家: {country_code}
    天气: {weather_desc}
    温度: {temp} {temp_unit}
    体感温度: {feels_like} {temp_unit}
    湿度: {humidity}%
    风速: {wind_speed} {wind_speed_unit}
    """

@mcp.tool()
async def get_weather(city: str, country_code: str = '', state_code: str = '', units: str = "metric", lang: str = "zh_cn") -> str:
    """获取指定城市的当前天气信息，如果用户使用中文询问某城市天气，你必须将城市名转换为相应的英文名称再调用API。

    Args:
        city: 城市名称 (例如：Beijing, Shanghai, New York)
        country_code: 国家代码(可选，例如：CN, US, JP)
        state_code: 州/省代码(可选，例如：BJ, SH, NY)
        units: 测量单位(可选，默认metric)
        lang: 语言(可选，默认zh_cn)
    
    Returns:
        str: 格式化的当前天气信息文本
    """
    
    # 构建位置查询参数，支持城市名称、州代码和国家代码组合
    location_query = city
    if state_code and country_code:
        location_query = f"{city},{state_code},{country_code}"    # 格式：城市，州代码，国家代码
    if country_code:
        location_query = f"{city},{country_code}"                 # 格式：城市，国家代码
    
    # 构建API请求URL,参数包含位置查询、API密钥、测量单位和语言
    url = f"{OPENWEATHER_API_BASE}/weather?q={location_query}&appid={OPENWEATHER_API_KEY}&units={units}&lang={lang}"
    data = await make_weather_request(url)      # 发送请求并获取响应数据
    if "cod" in data and data["cod"] != 200:    # 检查 API 响应中的错误代码
        return f"获取天气信息失败，错误：{data.get('message', '未知错误')}"
    
    return format_weather_data(data, units)

@mcp.tool()
async def get_forecast(city: str, country_code: str = None, state_code: str = None, units: str = "metric", lang: str = "zh_cn") -> str:
    """获取指定城市的5天天气预报信息，如果用户使用中文询问某城市5天天气预报，你必须将城市名转换为相应的英文名称再调用API。
    
    Args:
        city: 城市名称 (例如：Beijing, Shanghai, New York)
        country_code: 国家代码(可选，例如：CN, US, JP)
        state_code: 州/省代码(可选，例如：BJ, SH, NY)
        units: 测量单位(可选，默认metric)
        lang: 语言(可选，默认zh_cn)
    
    Returns:
        str: 格式化的5天天气预报信息文本
    """
    
    # 构建位置查询参数，支持城市名称、州代码和国家代码组合
    location_query = city
    if state_code and country_code:
        location_query = f"{city},{state_code},{country_code}"    # 格式：城市，州代码，国家代码
    if country_code:
        location_query = f"{city},{country_code}"                 # 格式：城市，国家代码
    
    # 构建API请求URL,参数包含位置查询、API密钥、测量单位和语言
    url = f"{OPENWEATHER_API_BASE}/forecast?q={location_query}&appid={OPENWEATHER_API_KEY}&units={units}&lang={lang}"
    data = await make_weather_request(url)                        # 发送请求并获取响应数据
    if not data:
        return "无法获取天气数据"
    if "cod" in data and data["cod"] != "200":                    # 检查 API 响应中的错误代码
        return f"获取5天天气预报信息失败，错误：{data.get('message', '未知错误')}"
    forecast_list = data.get("list", [])            # 提取5天天气预报信息
    if not forecast_list:
        return "无法获取5天天气预报信息"
    
    forecast_data = []
    # 遍历5天天气预报信息,并提取日期和时间
    for forecast in forecast_list:
        date_time = forecast.get("dt_txt", "")                                      # 提取日期和时间
        weather_desc = forecast.get("weather", [{}])[0].get("description", "未知")   # 提取天气描述
        temp = forecast.get("main", {}).get("temp", "未知")                          # 提取温度
        humidity = forecast.get("main", {}).get("humidity", "未知")                  # 提取湿度
        wind_speed = forecast.get("wind", {}).get("speed", "未知")                   # 提取风速
        
        # 根据 units 参数确定温度和风速的单位
        temp_unit = "°C" if units == "metric" else "°F"
        wind_speed_unit = "m/s" if units == "metric" else "mph"
        
        # 构建格式化的5天天气预报信息字符串
        forecast_str = f"""
        日期: {date_time}
        天气: {weather_desc}
        温度: {temp} {temp_unit}
        湿度: {humidity}%
        风速: {wind_speed} {wind_speed_unit}
        """
        
        forecast_data.append(forecast_str)

    return "\n---\n".join(forecast_data)

@mcp.prompt()
async def weather_prompt(city: str, weather_desc: str, temp: float, humidity: float, wind_speed: float, temp_unit: str, speed_unit: str) -> str:
    """用于生成天气报告的提示模板

    Args:
        city: 城市名称
        weather_desc: 天气描述
        temp: 温度
        humidity: 湿度
        wind_speed: 风速
        temp_unit: 温度单位
        speed_unit: 风速单位
    """
    
    return f"""请你作为专业的气象播报员，根据以下天气数据生成一份简介、易懂的天气报告:

    城市: {city}
    天气: {weather_desc}
    温度: {temp} {temp_unit}
    湿度: {humidity}%
    风速: {wind_speed} {speed_unit}

    报告内容包括：
    1.今日天气概况
    2.根据温度和湿度分析体感情况
    3.根据天气状况提供穿衣建议
    4.适合的户外活动推荐

    最后请使用自然、专业的语言，避免过于技术性的术语，要贴近生活。
    """

@mcp.tool()
async def weather_report(city: str, country_code: str = None, state_code: str = None, units: str = "metric", lang: str = "zh_cn") -> dict:
    """获取指定城市的天气信息并提供报告模板

    Args:
        city: 城市名称
        country_code: 国家代码(可选，例如：CN, US, JP)
        state_code: 州/省代码(可选，例如：BJ, SH, NY)
        units: 测量单位(可选，默认metric)
        lang: 语言(可选，默认zh_cn)
    
    Returns:
        dict: 包含天气数据和提示模板信息的字典
    """
    
    weather_result = await get_weather(city, country_code, state_code, units, lang)    # 获取原始天气信息

    # 使用正则表达式提取天气信息
    city_match = re.search(r'城市: (.*?)(?:\n|$)', weather_result)
    weather_match = re.search(r'天气: (.*?)(?:\n|$)', weather_result)
    temp_match = re.search(r'温度: ([\d.]+)(.*?)(?:\n|$)', weather_result)
    humidity_match = re.search(r'湿度: ([\d.]+)%', weather_result)
    wind_match = re.search(r'风速: ([\d.]+) (.*?)(?:\n|$)', weather_result)

    # 提取数据值，如果无法提取则使用默认值
    city_name = city_match.group(1) if city_match else city
    weather_desc = weather_match.group(1) if weather_match else "未知"
    temp_value = float(temp_match.group(1)) if temp_match else 0.0
    temp_unit = temp_match.group(2) if temp_match and len(temp_match.groups()) > 1 else "°C"
    humidity_value = int(float(humidity_match.group(1))) if humidity_match else 0
    wind_speed = float(wind_match.group(1)) if wind_match else 0.0
    speed_unit = wind_match.group(2) if wind_match and len(wind_match.groups()) > 1 else "m/s"

    return {
        "raw_data": weather_result,
        "prompt_template": "weather_prompt",
        "template_args": {
            "city": city_name,
            "weather_desc": weather_desc,
            "temp": temp_value,
            "temp_unit": temp_unit,
            "humidity": humidity_value,
            "wind_speed": wind_speed,
            "speed_unit": speed_unit,
        }
    }


if __name__ == "__main__":
    # Initialize and run the server
    mcp.run(transport='streamable-http')     # streamable-http：流式 http；stdio：本地标准传输；sse：服务器推送；

MCP Inspector

streamable-http传输模式

• 先单独启动服务，获取服务所在端口号：uv run weather.py

• 再启动调试器，输入对应的地址和端口：npx @modelcontextprotocol/inspector
• 注意：URL 那里 的 IP需要 0.0.0.0 而不是 localhost（可能我的本机使用了机场导致无法识别 localhost），最后是 mcp

stdio 传输模式

MCP 客户端

学习资源

• ‣

• ‣

要点总结

• 需要在项目下的 .env 文件中统一管理环境变量

• 关键组件解释
1. 客户端初始化：类初始化时包含会话管理、API 客户端、资源管理、交互管理
2. 服务器连接：支持 Python 和 Node.js 服务器
3. 查询处理：客户端与服务器的交互，从服务器返回对应的 tool 返回内容
4. 交互式界面：处理用户输入并显示响应、包含基本的错误处理
5. 资源管理：正确清理资源、连接问题错误处理、优雅关闭流程

代码示例

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

load_dotenv()    # 加载 .env 文件中的环境变量


# 定义 MCP 客户端类
class DeepSeekMCPClient:
    """一方面与MCP服务器进行通信，另一方面与DeepSeek API进行通信，帮助用户通过自然语言来使用各种强大的工具"""
    def __init__(self):
        self.session: Optional[ClientSession] = None       # 用于与MCP服务器通信的会话，初始值为 None
        self.exit_stack = AsyncExitStack()                 # 创建异步资源管理器，用于管理、释放资源，负责MCP 服务通信时接收和发送通信数据
        
        # 从环境变量获取信息
        self.llm_client = OpenAI(
            api_key=os.getenv("API_KEY"),
            base_url=os.getenv("BASE_URL")
        )
        self.model = os.getenv("MODEL")
    
    async def connect_to_server(self, server_script_path: str):
        """建立与 MCP 服务器的连接，根据服务脚本的类型(Python 或 JavaScript)选择正确的命令启动服务器，然后与之建立通信。
        参数: server_script_path: MCP 服务脚本路径，支持 Python(.py) 或 Node.js(.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                                       # 环境变量, 使用 None 表示继承当前环境变量
        )
        
        # 创建标准输入输出通信信道
        stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))       # 创建通信信道
        self.stdio, self.write = stdio_transport           # 解构对象中的读写通信，分别用于向MCP服务接收和发送数据
        self.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))   # 创建MCP客户端会话
        await self.session.initialize()                    # 初始MCP客户端会话，准备好与MCP服务进行通信
        response = await self.session.list_tools()         # 获取 MCP 服务提供的工具列表
        tools = response.tools                             # 获取工具列表
        
        print("\n已连接到MCP服务，可用的工具列表:", [tool.name for tool in tools])

    async def process_query(self, query: str) -> str:
        """处理用户查询，先用大模型判断使用哪些工具，然后调用这些工具，最后返回最终的回答。"""
        
        # 创建消息列表，用于存储用户的问题和模型的回答
        messages = [
            {
                "role": "system",
                "content": "你是一个专业的助手，可以通过调用合适的工具来帮助用户解决问题，请根据用户的需求选择最合适的工具。"
            },
            {
                "role": "user",
                "content": query
            }
        ]
        
        response = await self.session.list_tools()         # 请求 MCP 服务获取服务提供的工具列表
        tools = response.tools
        available_tools = [{
            "type": "function",                            # 工具类型，表示这是一个函数工具
            "function": {
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.inputSchema             # 工具参数
            }
        } for tool in tools]                               # 把工具信息转换成大模型需要的格式
        print(f"当前 MCP 服务所有工具列表: {available_tools}\n--------------------------------\n")
        
        # 大模型判断调用的工具
        response = self.llm_client.chat.completions.create(
            model=self.model,
            messages=messages,
            tools=available_tools if available_tools else None,
            temperature=0.5,
            max_tokens=4096)
        reply = response.choices[0].message               # 获取模型的回答；包含 role、content、tool_calls
        messages.append(reply)                            # 添加到历史消息中，确保模型记忆

        final_text = []
        # 检查模型响应中是否包含工具调用请求，如果有则调用工具
        if hasattr(reply, "tool_calls") and reply.tool_calls:
            for tool_call in reply.tool_calls:
                tool_name = tool_call.function.name
                try:
                    tool_args = json.loads(tool_call.function.arguments)   # 解析参数
                except json.JSONDecodeError:
                    tool_args = {}
                final_text.append(f"调用工具: {tool_name}, 参数: {tool_args}\n")
                print(f"准备调用工具: {tool_name} 参数: {tool_args}\n--------------------------------\n")
                result = await self.session.call_tool(tool_name, tool_args)  # 异步调用 MCP 服务上的工具
                print(f"工具 {tool_name} 执行结果: {result}\n--------------------------------\n")

                # 确保工具结果是字符串格式
                tool_result_content = result.content
                if isinstance(tool_result_content, list):        # 如果工具结果是列表，则将列表中的每个元素转换为字符串并添加到最终文本中
                    text_content = ""
                    for item in tool_result_content:
                        if hasattr(item, 'text'):
                            text_content += item.text
                    tool_result_content = text_content
                elif not isinstance(tool_result_content, str):   # 如果不是字符串，则转换为字符串
                    tool_result_content = str(tool_result_content)
                print(f"工具返回结果(格式化后): {tool_result_content}\n--------------------------------\n")

                # 将工具调用结果添加到历史消息中，保证与模型会话的连贯性
                tool_message = {
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": tool_result_content,
                }
                messages.append(tool_message)

                # 尝试解析工具返回的JSON结果，检查是否包含MCP模板结构
                try:
                    tool_result_json = json.loads(tool_result_content)
                    # 检查是否包含 MCP 模板结构(具有 prompt_template 和 template_args 字段)
                    if(isinstance(tool_result_json, dict) and "prompt_template" in tool_result_json and "template_args" in tool_result_json):
                        raw_data = tool_result_json["raw_data"]
                        prompt_template = tool_result_json["prompt_template"]
                        template_args = tool_result_json["template_args"]

                        # 将模板参数转换为字符串类型(MCP规范要求)
                        string_args = {k:str(v) for k,v in template_args.items()}
                        print(f"模板名称: {prompt_template}, 模板参数: {string_args}\n--------------------------------\n")

                        # 调用 MCP 服务上的工具，传入工具名称和函数参数，返回工具函数执行结果
                        template_response = await self.session.get_prompt(prompt_template, string_args)
                        print(f"模板响应: {template_response}\n--------------------------------\n")

                        if hasattr(template_response, "messages") and template_response.messages:
                            print(f"模板具体的信息: {template_response.messages}\n--------------------------------\n")
                            for msg in template_response.messages:
                                content = msg.content.text if hasattr(msg.content, "text") else msg.content
                                template_message = {
                                    "role": msg.role,            # 保持原始角色
                                    "content": content           # 消息内容
                                }
                                print(f"模板消息历史: {template_message}\n--------------------------------\n")
                                messages.append(template_message)
                        else:
                            print("警告：模板响应中没有包含消息内容。")
                except json.JSONDecodeError:
                    pass
                
                # 再次调用模型根据工具结果生成最终的回答
                try:
                    print("正在请求 DeepSeek API 生成最终回答...")
                    # 发送包含工具调用和结果的完整消息历史
                    final_response = self.llm_client.chat.completions.create(
                        model=self.model,
                        messages=messages,
                        temperature=0.5,
                        max_tokens=4096)
                    final_content = "DeepSeek回答：" + final_response.choices[0].message.content
                    if final_content:
                        final_text.append(final_content)
                    else:
                        print("警告：DeepSeek API 没有生成任何内容。")
                        final_text.append(f"工具调用结果：\n{tool_result_content}")
                except Exception as e:
                    print(f"生成最终回复时出错: {e}")
                    final_text.append(f"工具返回结果：\n{tool_result_content}")
        else:
            if reply.content:
                final_text.append(f"{reply.content}")
            else:
                final_text.append("模型没有生成有效回复。")
        
        return '\n'.join(final_text)
    
    async def chat_loop(self):
        """
        运行交互式聊天循环，处理用户输入并显示回复；不断接收用户输入，处理问题，并显示回答，直到用户输入'quit'退出。
        """
        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 = DeepSeekMCPClient()                             # 创建客户端实例
    
    try:
        await client.connect_to_server(sys.argv[1])           # 连接到MCP服务器
        await client.chat_loop()                              # 启动聊天循环
    finally:
        await client.cleanup()                                # 清理资源


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

# uv run mcp-weather-client/client.py /Users/ayd/Desktop/MCP/res/mcp-weather-server/weather.py

其他相关资源站

官方资源

• 官方文档：‣

• 官方 MCP 服务器

• 官方 sdk：

MCP 客户端

• ‣

• ‣

MCP 服务器

• ‣

• ‣

• ‣

• ‣

• ‣

开发工具

• 第三方 MCP 开发工具：

学习视频

• B 站尚硅谷：‣

• B 站Hucci写代码：‣