当前位置：首页 > news >正文

MCP入门实战（Python版）

news 来源：原创 2025/6/22 8:01:45

MCP介绍

MCP入门介绍

MCP 简介 - MCP 中文文档

MCP，全称是Model Context Protocol，模型上下文协议，由Claude母公司Anthropic于2024年11月正式提出。

从本质上来说，MCP是一种技术协议，一种智能体Agent开发过程中共同约定的一种规范。这就好比秦始皇的“书同文、车同轨”，在统一的规范下，大家的协作效率就能大幅提高，最终提升智能体Agent的开发效率。截止目前，已上千种MCP工具诞生，在强悍的MCP生态加持下，人人手搓Manus的时代即将到来。

总的来说，MCP解决的最大痛点，就是Agent开发中调用外部工具的技术门槛过高的问题。

我们都知道，能调用外部工具，是大模型进化为智能体Agent的关键，如果不能使用外部工具，大模型就只能是个简单的聊天机器人，甚至连查询天气都做不到。由于底层技术限制，大模型本身是无法和外部工具直接通信的，因此Function calling的思路，就是创建一个外部函数（function）作为中介，一边传递大模型的请求，另一边调用外部工具，最终让大模型能够间接的调用外部工具。

例如，当我们要查询当前天气时，让大模型调用外部工具的function calling的过程就如图所示：

Function calling是个非常不错的技术设计，自诞生以来，一直被业内奉为圭臬。但唯一的问题就是，编写这个外部函数的工作量太大了，一个简单的外部函数往往就得上百行代码，而且，为了让大模型“认识” 这些外部函数，我们还要额外为每个外部函数编写一个JSON Schema格式的功能说明，此外，我们还需要精心设计一个提示词模版，才能提高Function calling响应的准确率。

而MCP的目标，就是能在Agent开发过程中，让大模型更加便捷的调用外部工具。为此，MCP提出了两个方案

其一，“车同轨、书同文”，统一Function calling的运行规范。首先是先统一名称，MCP把大模型运行环境称作 MCP Client，也就是MCP客户端，同时，把外部函数运行环境称作MCP Server，也就是MCP服务器
然后，统一MCP客户端和服务器的运行规范，并且要求MCP客户端和服务器之间，也统一按照某个既定的提示词模板进行通信。

“车同轨、书同文”最大的好处就在于，可以避免MCP服务器的重复开发，也就是避免外部函数重复编写。例如，像查询天气、网页爬取、查询本地MySQL数据库这种通用的需求，大家有一个人开发了一个服务器就好，开发完大家都能复制到自己的项目里来使用，不用每个人每次都单独写一套。现在，只要你本地运行的大模型支持MCP协议，也就是只要安装了相关的库，仅需几行代码即可接入这些海量的外部工具。

这种“车同轨、书同文”的规范，在技术领域就被称作协议，例如http就是网络信息交换的技术协议。各类技术协议的目标，都是希望通过提高协作效率来提升开发效率，而MCP，Model Context Protocol，就是一种旨在提高大模型Agent开发效率的技术协议。

那既然是协议，必然是使用的人越多才越有用。因此，为了进一普及MCP协议，Anthropic还提供了一整套MCP客户端、服务器开发的SDK，也就是开发工具，并且支持Python、TS和Java等多种语言，借助 SDK，仅需几行代码，就可以快速开发一个MCP服务器。

# server.py
from mcp.server.fastmcp import FastMCP# Create an MCP server
mcp = FastMCP("Demo")# Add an addition tool
@mcp.tool()
def add(a: int, b: int) -> int:"""Add two numbers"""return a + b# Add a dynamic greeting resource
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:"""Get a personalized greeting"""return f"Hello, {name}!"

然后，你就可以把它接入任意一个MCP客户端来构建智能体，如果愿意，还可以把MCP服务器分享到社区，给有需求的开发者使用，甚至你还可以把你的MCP服务器放到线上运行，让用户付费使用。

而MCP的客户端，不仅支持Claude模型，也支持任意本地模型或者在线大模型，或者是一些IDE。例如，现在Cursor正式接入MCP，代表着Cursor正式成为MCP客户端，在Cursor中，我们不仅能快速编写 MCP服务器（外部函数），更能借助Cursor一键连接上成百上千的开源MCP服务器，让大模型快速接入海量工具，从而大幅加快Agent开发进度。

Function calling技术简介

MCP客户端开发

UV工具入门使用指南

UV介绍

MCP开发要求借助uv进行虚拟环境创建和依赖管理。

uv 是一个Python 依赖管理工具，类似于 pip 和 conda ，但它更快、更高效，并且可以更好地管理 Python 虚拟环境和依赖项。它的核心目标是替代 pip 、venv 和 pip-tools ，提供更好的性能和更低的管理开销。

uv 的特点：

1. 速度更快：相比 pip ，uv 采用 Rust 编写，性能更优。
2. 支持 PEP 582：无需 virtualenv ，可以直接使用
3. 兼容 pip ：支持 __pypackages__ 进行管理。 requirements.txt 和 pyproject.toml 依赖管理。
4. 替代 venv ：提供 uv venv 进行虚拟环境管理，比venv 更轻量。
5. 跨平台：支持 Windows、macOS 和 Linux。

uv安装流程

方法 1：使用 pip 安装（适用于已安装 pip install uv venv 更轻量。 pip 的系统）

pip install uv

方法 2：使用 curl 直接安装如果你的系统没有 pip ，可以直接运行：

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

这会自动下载 uv 并安装到 /usr/local/bin 。

uv的基本用法介绍

安装 uv 后，你可以像 pip 一样使用它，但它的语法更简洁，速度也更快。注意，以下为使用语法示例，不用实际运行。

安装 Python 依赖

uv pip install requests

与 pip install requests 类似，但更快

创建虚拟环境

uv venv myenv

等效于 python -m venv myenv，但更高效。

激活虚拟环境

source myenv/bin/activate  # Linux/macOS
myenv\Scripts\activate     # Windows

安装 requirements.txt

 uv pip install -r requirements.txt

直接运行 Python 项目

如果项目中包含 pyproject.toml ，你可以直接运行：

uv run python script.py

这等效于：

pip install -r requirements.txt
python script.py

但 uv 速度更快，管理更高效。

为什么MCP更推荐使用uv进行环境管理？

MCP 依赖的 Python 环境可能包含多个模块，uv 通过 pyproject.toml 提供更高效的管理方式，并且可以避免 pip 的一些依赖冲突问题。此外，uv 的包管理速度远超 pip ，这对于 MCP 这样频繁管理依赖的项目来说是一个很大的优势。

接下来我们尝试先构建一个 MCP 客户端，确保基本逻辑可用，然后再逐步搭建 MCP 服务器进行联调，这样可以分阶段排查问题，避免一上来就涉及太多复杂性。

MCP极简客户端搭建流程

前文我们已经说过，MCP把大模型运行环境称作 MCP Client，也就是MCP客户端，同时，把外部函数运行环境称作MCP Server，也就是MCP服务器

因此创建MCP客户端的流程其实就是创建一个调用大模型的接口

创建MCP 客户端项目

uv init mcp-client
cd mcp-client

创建MCP客户端虚拟环境

# 创建虚拟环境
uv venv
# 激活虚拟环境
source .venv/bin/activate

这里需要注意的是，相比pip，uv会自动识别当前项目主目录并创建虚拟环境。

然后即可通过add方法在虚拟环境中安装相关的库。

# 安装 MCP SDK
uv add mcp

为了支持调用OpenAI模型，以及在环境变量中读取API-KEY等信息，需要先安装如下依赖：

uv add mcp openai python-dotenv

创建.env文件

接下来创建.env文件，并写入OpenAI的API-Key，以及反向代理地址。借助反向代理，国内可以无门槛直连OpenAI官方服务器，并调用官方API。

使用线上模型

BASE_URL="反向代理地址"
MODEL=gpt-4o
OPENAI_API_KEY="OpenAI-API-Key"

而如果是使用DeepSeek模型，则需要在.env中写入如下内容

BASE_URL=https://api.deepseek.com
MODEL=deepseek-chat      
OPENAI_API_KEY="DeepSeek API-Key"

使用本地部署模型

如果是本地部署的模型，比如我自己使用ollama本地部署的qwen3模型，则使用以下配置：

BASE_URL=http://localhost:11434/v1/
MODEL=qwen3:1.7b
OPENAI_API_KEY=ollama

如果使用的是VLLM部署的模型，则使用以下配置

BASE_URL=http://localhost:8000/v1
MODEL=./QwQ-32B
OPENAI_API_KEY=EMPTY

本地部署的模型调用过程中，api_key其实并不起作用，因此随便填写即可

关于如何本地部署qwen3模型，可参考

零基础本地部署Qwen3模型（ollama+Open-WebUI）-CSDN博客

vLLM调度部署Qwen3-CSDN博客

创建client.py

import asyncio
import os
from openai import OpenAI
from dotenv import load_dotenv
from contextlib import AsyncExitStack# 加载 .env 文件，确保 API Key 受到保护
load_dotenv()class MCPClient:def __init__(self):"""初始化 MCP 客户端"""self.exit_stack = AsyncExitStack()self.openai_api_key = os.getenv("OPENAI_API_KEY")  # 读取 OpenAI API Keyself.base_url = os.getenv("BASE_URL")  # 读取 BASE URLself.model = os.getenv("MODEL")  # 读取 modelif not self.openai_api_key:raise ValueError("❌ 未找到 OpenAI API Key，请在 .env 文件中设置 OPENAI_API_KEY")self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url)async def process_query(self, query: str) -> str:"""调用 OpenAI API 处理用户查询"""messages = [{"role": "system", "content": "你是一个智能助手，帮助用户回答问题。"},{"role": "user", "content": query}]try:# 调用 OpenAI APIresponse = await asyncio.get_event_loop().run_in_executor(None,lambda: self.client.chat.completions.create(model=self.model,messages=messages))return response.choices[0].message.contentexcept Exception as e:return f"⚠ 调用 OpenAI API 时出错: {str(e)}"async def chat_loop(self):"""运行交互式聊天循环"""print("\n🤖MCP 客户端已启动！输入 'quit' 退出")while True:try:query = input("\n你: ").strip()if query.lower() == 'quit':break# 发送用户输入到 OpenAI APIresponse = await self.process_query(query)print(f"\n🤖 OpenAI: {response}")except Exception as e:print(f"\n⚠ 发生错误: {str(e)}")async def cleanup(self):"""清理资源"""await self.exit_stack.aclose()async def main():client = MCPClient()try:await client.chat_loop()finally:await client.cleanup()if __name__ == "__main__":asyncio.run(main())

运行

使用线上模型运行

如果客户端client.py使用的是线上模型，则直接运行代码即可

uv run client.py

使用本地部署模型运行

如果客户端使用的是本地部署的模型，则需要首先开启本地部署的模型服务，这样client.py才能找到要调用的模型。使用本地部署模型的工具有多种，如ollama、VLLM等

如果我们使用ollama部署的本地模型，则只需输入以下命令即可启动我们在本地部署的模型服务

ollama start

如果我们使用VLLM部署的本地模型，则输入以下命令启动部署的模型服务

 CUDA_VISIBLE_DEVICES=0,1 vllm serve ./QwQ-32B --tensor-parallel-size 2

然后再重新运行我们的 client.py

uv run client.py

运行效果如下所示

MCP服务端开发

MCP服务器概念介绍

根据MCP协议定义，Server可以提供三种类型的标准能力，Resources、Tools、Prompts，每个 Server可同时提供者三种类型能力或其中一种。

Resources：资源，类似于文件数据读取，可以是文件资源或是API响应返回的内容。
Tools：工具，第三方服务、功能函数，通过此可控制LLM可调用哪些函数。
Prompts：提示词，为用户预先定义好的完成特定任务的模板。

MCP服务器通讯机制

Model Context Protocol（MCP）是一种由 Anthropic 开源的协议，旨在将大型语言模型直接连接至数据源，实现无缝集成。

根据 MCP 的规范，当前支持两种传输方式：标准输入输出（stdio）和基于 HTTP 的服务器推送事件（SSE），分别对应着本地与远程通讯。Client与Server间使用JSON-RPC 2.0格式进行消息传输。

本地通讯：使用了stdio传输数据，具体流程Client启动Server程序作为子进程，其消息通讯是通过 stdin/stdout进行的，消息格式为JSON-RPC 2.0。
远程通讯：Client与Server可以部署在任何地方，Client使用SSE与Server进行通讯，消息的格式为 JSON-RPC 2.0，Server定义了/see与/messages接口用于推送与接收数据。

而近期，开发者在 MCP 的 GitHub 仓库中提交了一项提案，建议采用 “可流式传输的 HTTP”来替代现有的 HTTP+SSE 方案。此举旨在解决当前远程 MCP 传输方式的关键限制，同时保留其优势。 HTTP 和 SSE（服务器推送事件）在数据传输方式上存在明显区别：

通信方式：
HTTP：采用请求-响应模式，客户端发送请求，服务器返回响应，每次请求都是独立的。
SSE：允许服务器通过单个持久的 HTTP 连接，持续向客户端推送数据，实现实时更新。

连接特性：
HTTP：每次请求通常建立新的连接，虽然在 HTTP/1.1 中引入了持久连接，但默认情况下仍是短连接。
SSE：基于长连接，客户端与服务器之间保持持续的连接，服务器可以在任意时间推送数据。

适用场景：
HTTP：适用于传统的请求-响应场景，如网页加载、表单提交等。
SSE：适用于需要服务器主动向客户端推送数据的场景，如实时通知、股票行情更新等。需要注意的是，SSE 仅支持服务器向客户端的单向通信，而 WebSocket 则支持双向通信。

需要注意的是，SSE 仅支持服务器向客户端的单向通信，而 WebSocket 则支持双向通信。

天气查询服务器Server创建

准备工作

为方便，我们尝试一个入门级的示例，那就是创建一个天气查询的服务器。通过使用OpenWeather API，创建一个能够实时查询天气的服务器（server），并使用stdio方式进行通信。

为了使用天气查询API，我们首先需要去OpenWeather注册并获取自己的api_key

OpenWeather官网：https://openweathermap.org/

注册成功之后即可获取自己的api_key

申请完毕后可使用以下代码进行api测试

curl -s "https://api.openweathermap.org/data/2.5/weather?q=Beijing&appid=${api-key}&units=metric&lang=zh_cn"

准备工作做好之后，接下来我们正式进入MCP server的开发

服务器依赖安装

由于我们需要使用http请求来查询天气，因此需要在当前虚拟环境中添加如下依赖

uv add mcp httpx

服务器代码编写

接下来尝试创建服务器代码，此时MCP基本执行流程如下：

对应server服务器代码如下：

import json
import httpx
from typing import Any
from mcp.server.fastmcp import FastMCP# 初始化 MCP 服务器
mcp = FastMCP("WeatherServer")# OpenWeather API 配置
OPENWEATHER_API_BASE = "https://api.openweathermap.org/data/2.5/weather"
# API_KEY = "YOUR_API_KEY"  # 请替换为你自己的 OpenWeather API Key
API_KEY = "62f016856f6a66a0cd6ee3cda5dbc9c3"  
USER_AGENT = "weather-app/1.0"async def fetch_weather(city: str) -> dict[str, Any] | None:"""从 OpenWeather API 获取天气信息。:param city: 城市名称（需使用英文，如 Beijing）:return: 天气数据字典；若出错返回包含 error 信息的字典"""params = {"q": city,"appid": API_KEY,"units": "metric","lang": "zh_cn"}headers = {"User-Agent": USER_AGENT}async with httpx.AsyncClient() as client:try:response = await client.get(OPENWEATHER_API_BASE, params=params, headers=headers, timeout=30.0)response.raise_for_status()return response.json()  # 返回字典类型except httpx.HTTPStatusError as e:return {"error": f"HTTP 错误: {e.response.status_code}"}except Exception as e:return {"error": f"请求失败: {str(e)}"}def format_weather(data: dict[str, Any] | str) -> str:"""将天气数据格式化为易读文本。:param data: 天气数据（可以是字典或 JSON 字符串）:return: 格式化后的天气信息字符串"""# 如果传入的是字符串，则先转换为字典if isinstance(data, str):try:data = json.loads(data)except Exception as e:return f"无法解析天气数据: {e}"# 如果数据中包含错误信息，直接返回错误提示if "error" in data:return f"⚠ {data['error']}"# 提取数据时做容错处理city = data.get("name", "未知")country = data.get("sys", {}).get("country", "未知")temp = data.get("main", {}).get("temp", "N/A")humidity = data.get("main", {}).get("humidity", "N/A")wind_speed = data.get("wind", {}).get("speed", "N/A")# weather 可能为空列表，因此用 [0] 前先提供默认字典weather_list = data.get("weather", [{}])description = weather_list[0].get("description", "未知")return (f"🌍 {city}, {country}\n"f"🌡 温度: {temp}°C\n"f"💧 湿度: {humidity}%\n"f"🌬 风速: {wind_speed} m/s\n"f"⛅ 天气: {description}\n")@mcp.tool()
async def query_weather(city: str) -> str:"""输入指定城市的英文名称，返回今日天气查询结果。:param city: 城市名称（需使用英文）:return: 格式化后的天气信息"""data = await fetch_weather(city)return format_weather(data)if __name__ == "__main__":# 以标准 I/O 方式运行 MCP 服务器mcp.run(transport='stdio')

天气查询客户端client创建

前边我们创建一个极简的客户端代码，本质上就是创建一个调用大模型的代码，现在我们重新修改客户端代码，以支持MCP协议

import asyncio
import os
import json
import sys
from typing import Optional
from contextlib import AsyncExitStack
from openai import OpenAI
from dotenv import load_dotenv
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client# 加载 .env 文件，确保 API Key 受到保护
load_dotenv()class MCPClient:def __init__(self):"""初始化 MCP 客户端"""self.exit_stack = AsyncExitStack()self.openai_api_key = os.getenv("OPENAI_API_KEY")  # 读取 OpenAI API Keyself.base_url = os.getenv("BASE_URL")  # 读取 BASE URLself.model = os.getenv("MODEL")  # 读取 modelif not self.openai_api_key:raise ValueError("❌ 未找到 OpenAI API Key，请在 .env 文件中设置 OPENAI_API_KEY")self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url)self.session: Optional[ClientSession] = Noneself.exit_stack = AsyncExitStack()async def connect_to_server(self, server_script_path: str):print(server_script_path)"""连接到 MCP 服务器并列出可用工具"""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)# 启动 MCP 服务器并建立通信stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))self.stdio, self.write = stdio_transportself.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))await self.session.initialize()# 列出 MCP 服务器上的工具response = await self.session.list_tools()tools = response.toolsprint("\n已连接到服务器，支持以下工具:", [tool.name for tool in tools])async def process_query(self, query: str) -> str:"""使用大模型处理查询并调用可用的 MCP 工具 (Function Calling)"""messages = [{"role": "user", "content": query}]response = await self.session.list_tools()available_tools = [{"type": "function","function": {"name": tool.name,"description": tool.description,"input_schema": tool.inputSchema}} for tool in response.tools]# print(available_tools)response = self.client.chat.completions.create(model=self.model,messages=messages,tools=available_tools)# 处理返回的内容content = response.choices[0]if content.finish_reason == "tool_calls":# 如果需要使用工具，就解析工具tool_call = content.message.tool_calls[0]tool_name = tool_call.function.nametool_args = json.loads(tool_call.function.arguments)# 执行工具result = await self.session.call_tool(tool_name, tool_args)print(f"\n\n[Calling tool {tool_name} with args {tool_args}]\n\n")# 将模型返回的调用哪个工具数据和工具执行完成后的数据都存入messages中messages.append(content.message.model_dump())messages.append({"role": "tool","content": result.content[0].text,"tool_call_id": tool_call.id,})# 将上面的结果再返回给大模型用于生产最终的结果response = self.client.chat.completions.create(model=self.model,messages=messages,)return response.choices[0].message.contentreturn content.message.contentasync def chat_loop(self):"""运行交互式聊天循环"""print("\n🤖 MCP 客户端已启动！输入 'quit' 退出")while True:try:query = input("\n你: ").strip()if query.lower() == 'quit':breakresponse = await self.process_query(query)  # 发送用户输入到 OpenAI APIprint(f"\n🤖 OpenAI: {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("Usage: python client.py <path_to_server_script>")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 sysasyncio.run(main())

代码解读

导入必要库

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

asyncio ：支持异步编程
os / json ：读取环境变量、解析 JSON
typing.Optional ：类型提示
contextlib.AsyncExitStack ：用于安全管理异步资源（如 MCP 连接）
openai.OpenAI ：你的自定义 OpenAI Client 类
dotenv.load_dotenv ：从 .env 文件加载环境变量（如 API Key）
MCP 相关： mcp.ClientSession , mcp.client.stdio , load_dotenv() StdioServerParameters

加载环境变量

从 .env 文件中加载环境变量

load_dotenv()

初始化工作

    def __init__(self):"""初始化 MCP 客户端"""self.exit_stack = AsyncExitStack()self.openai_api_key = os.getenv("OPENAI_API_KEY")  # 读取 OpenAI API Keyself.base_url = os.getenv("BASE_URL")  # 读取 BASE URLself.model = os.getenv("MODEL")  # 读取 modelif not self.openai_api_key:raise ValueError("❌ 未找到 OpenAI API Key，请在 .env 文件中设置 OPENAI_API_KEY")self.client = OpenAI(api_key=self.openai_api_key, base_url=self.base_url)self.session: Optional[ClientSession] = Noneself.exit_stack = AsyncExitStack()

1. self.exit_stack = AsyncExitStack()
用于统一管理异步上下文（如 MCP 连接）的生命周期。
可以在退出（ cleanup ）时自动关闭

2. 读取环境变量 openai_api_key ：
OpenAI API Key
base_url ：模型请求的 Base URL（如你自建的反代地址）
model ：OpenAI 模型名称

3. 初始化 OpenAI 客户端
OpenAI(api_key=self.openai_api_key, base_url=self.base_url) ：你自定义的 OpenAI 客户端，用来与 OpenAI Chat Completion API 通信。

4. self.session
用于保存 MCP 的客户端会话，默认是 None ，稍后通过 connect_to_server 进行连接。

5. 再次声明 self.exit_stack = AsyncExitStack() 这里两次赋值其实有点冗余（前面已赋值过一次）。不过并不影响功能，等同于覆盖掉前面的对象。可能是手误或调试时多写了一次

初始化MCP服务器连接

command = "python" if is_python else "node"server_params = StdioServerParameters(command=command,args=[server_script_path],env=None)# 启动 MCP 服务器并建立通信stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))self.stdio, self.write = stdio_transportself.session = await self.exit_stack.enter_async_context(ClientSession(self.stdio, self.write))await self.session.initialize()# 列出 MCP 服务器上的工具response = await self.session.list_tools()tools = response.toolsprint("\n已连接到服务器，支持以下工具:", [tool.name for tool in tools])

StdioServerParameters ：告诉 MCP 客户端如何启动服务器。
- command=command ：如 "python"
- args=[server_script_path] ：如 ["weather_server.py"]
stdio_client(server_params) ：启动服务器进程，并建立标准 I/O 通信管道。
self.stdio, self.write = stdio_transport ：拿到读写流。
ClientSession(...) ：创建 MCP 客户端会话，与服务器交互。
await self.session.initialize() ：发送初始化消息给服务器，等待服务器就绪。
list_tools() ：向 MCP 服务器请求所有已注册的工具（用 @mcp.tool() 标记）。

客户端模型调用MCP服务方法

    async def process_query(self, query: str) -> str:"""使用大模型处理查询并调用可用的 MCP 工具 (Function Calling)"""messages = [{"role": "user", "content": query}]

收到用户输入后，先把它组装进一个 messages 列表，目前只包含用户信息（ "content": query} ）。

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

获取服务器上的工具，再转换成 available_tools 的格式。
这里自定义了一个结构：每个工具对应一个 {"type": "function", "function": {...}} 的字典。方便后面发给 OpenAI，告诉它：可以调用这些工具。

       response = self.client.chat.completions.create(model=self.model,messages=messages,tools=available_tools)

使用 OpenAI 客户端的法发送请求：

model=self.model ：比如 "gpt-4o" 或"deepseek-chat"
messages=messages ：聊天上下文 "deepseek-chat"
tools=available_tools ：让模型知道有哪些可调用的「函数」。这是你自定义的 “Function Calling”协议（非官方 JSON schema）。

# 处理返回的内容content = response.choices[0]if content.finish_reason == "tool_calls":# 如果需要使用工具，就解析工具tool_call = content.message.tool_calls[0]tool_name = tool_call.function.nametool_args = json.loads(tool_call.function.arguments)# 执行工具result = await self.session.call_tool(tool_name, tool_args)print(f"\n\n[Calling tool {tool_name} with args {tool_args}]\n\n")# 将模型返回的调用哪个工具数据和工具执行完成后的数据都存入messages中messages.append(content.message.model_dump())messages.append({"role": "tool","content": result.content[0].text,"tool_call_id": tool_call.id,})# 将上面的结果再返回给大模型用于生产最终的结果response = self.client.chat.completions.create(model=self.model,messages=messages,)return response.choices[0].message.content

模型调用MCP服务并执行返回相关结果

运行

 uv run client-mcp.py server.py

以下是相关运行效果

Processing request of type ListToolsRequest已连接到服务器，支持以下工具: ['query_weather']🤖 MCP 客户端已启动！输入 'quit' 退出你: 介绍一下你自己
Processing request of type ListToolsRequest
[{'type': 'function', 'function': {'name': 'query_weather', 'description': '\n    输入指定城市的英文名称，返回今日天气查询结果。\n    :param city: 城市名称（需使用英文）\n    :return: 格式化后的天气信息\n    ', 'input_schema': {'properties': {'city': {'title': 'City', 'type': 'string'}}, 'required': ['city'], 'title': 'query_weatherArguments', 'type': 'object'}}}]🤖 OpenAI: <think>
好的，用户让我介绍一下自己。首先，我需要确认自己是否需要调用任何工具。根据提供的工具，有一个查询天气的函数，但用户的问题是关于自我介绍，不需要天气信息。因此，我应该直接回答，而不需要使用任何工具。接下来，我需要用简洁明了的方式说明自己的身份和能力，比如我是智能助手，能提供帮助等。确保回答友好且信息准确。
</think>我是智能助手，可以为您提供帮助。如果您有任何问题或需要 assistance，欢迎随时告诉我！你: 你目前有哪些功能
Processing request of type ListToolsRequest
[{'type': 'function', 'function': {'name': 'query_weather', 'description': '\n    输入指定城市的英文名称，返回今日天气查询结果。\n    :param city: 城市名称（需使用英文）\n    :return: 格式化后的天气信息\n    ', 'input_schema': {'properties': {'city': {'title': 'City', 'type': 'string'}}, 'required': ['city'], 'title': 'query_weatherArguments', 'type': 'object'}}}]🤖 OpenAI: <think>
嗯，用户问我现在有哪些功能。首先，我需要回顾一下提供的工具信息。用户提供的工具里只有一個函数，叫做query_weather，功能是查询指定城市的天气。这个函数的参数是城市名称，要求用英文。所以，我的功能主要是天气查询，只能处理城市名称的英文输入。接下来，用户可能想知道我的能力范围，或者是否能提供其他服务。但根据当前的工具，我只能处理天气相关的问题。我需要明确告诉用户我的功能限制，同时提到可以协助查询天气。可能用户想测试我的其他功能，但目前没有，所以应该如实说明。另外，用户可能希望知道是否需要提供其他帮助，比如天气预报或建议，但根据现有工具，只能处理天气查询。所以回答时要诚实，同时提供可能的帮助方向。比如，告诉用户我现在能查天气，需要输入城市名称的英文，然后举例说明，比如伦敦或纽约。这样用户能明白操作方法。
</think>目前我只能查询天气信息，功能包括：
1. 支持查询全球城市天气（需输入英文名称）
2. 返回格式化后的天气信息（如温度、湿度、风力等）
3. 时效性：今日天气数据请提供需要查询的城市英文名称，例如"London"或"New York"。你: 帮我查询北京的天气
Processing request of type ListToolsRequest
[{'type': 'function', 'function': {'name': 'query_weather', 'description': '\n    输入指定城市的英文名称，返回今日天气查询结果。\n    :param city: 城市名称（需使用英文）\n    :return: 格式化后的天气信息\n    ', 'input_schema': {'properties': {'city': {'title': 'City', 'type': 'string'}}, 'required': ['city'], 'title': 'query_weatherArguments', 'type': 'object'}}}]
Processing request of type CallToolRequest
HTTP Request: GET https://api.openweathermap.org/data/2.5/weather?q=Beijing&appid=62f016856f6a66a0cd6ee3cda5dbc9c3&units=metric&lang=zh_cn "HTTP/1.1 200 OK"[Calling tool query_weather with args {'city': 'Beijing'}]🤖 OpenAI: <think>
好的，用户让我查北京的天气，我之前已经用英文查询了，现在需要根据返回的数据整理成自然的中文回复。首先，确认返回的数据结构，温度是33.46°C，湿度22%，风速1.68 m/s，天气晴。需要把这些数据用中文表达清楚，同时保持口语化，避免使用Markdown格式。用户可能希望得到简洁明了的天气信息，不需要太技术性的术语，但要准确。比如“温度: 33.46°C”可以翻译成“温度：33.46°C”，保持数字格式。湿度和风速也要类似处理。最后加上天气状况，比如“晴”，确保信息完整。另外，用户可能没有明确说，但可能需要提醒注意防晒或补水，但根据问题本身，可能不需要添加额外信息。保持回答简洁，直接给出数据即可。检查是否有遗漏的参数，比如是否有风向，但返回中没有提到，所以不需要补充。确保单位正确，如°C、m/s等。最后用自然的中文句式连接各部分，比如“目前北京天气晴朗，温度为33.46°C，湿度22%，风速1.68 m/s。”这样用户就能清楚了解天气情况了。
</think>目前北京天气晴朗，温度为33.46°C，湿度22%，风速1.68 m/s。你:

需要注意的是，当我问大模型当前它有哪些功能的时候，它回答如下所示

这是因为，目前我们的MCP服务端只写了一个工具，即天气查询，接下来，我们可以尝试自己写一些服务工具来测试

自定义MCP服务工具

修改服务端代码

import json
import httpx
from typing import Any
from mcp.server.fastmcp import FastMCP# 初始化 MCP 服务器
mcp = FastMCP("WeatherServer")# OpenWeather API 配置
OPENWEATHER_API_BASE = "https://api.openweathermap.org/data/2.5/weather"
# API_KEY = "YOUR_API_KEY"  # 请替换为你自己的 OpenWeather API Key
API_KEY = "62f016856f6a66a0cd6ee3cda5dbc9c3"  
USER_AGENT = "weather-app/1.0"async def fetch_weather(city: str) -> dict[str, Any] | None:"""从 OpenWeather API 获取天气信息。:param city: 城市名称（需使用英文，如 Beijing）:return: 天气数据字典；若出错返回包含 error 信息的字典"""params = {"q": city,"appid": API_KEY,"units": "metric","lang": "zh_cn"}headers = {"User-Agent": USER_AGENT}async with httpx.AsyncClient() as client:try:response = await client.get(OPENWEATHER_API_BASE, params=params, headers=headers, timeout=30.0)response.raise_for_status()return response.json()  # 返回字典类型except httpx.HTTPStatusError as e:return {"error": f"HTTP 错误: {e.response.status_code}"}except Exception as e:return {"error": f"请求失败: {str(e)}"}def format_weather(data: dict[str, Any] | str) -> str:"""将天气数据格式化为易读文本。:param data: 天气数据（可以是字典或 JSON 字符串）:return: 格式化后的天气信息字符串"""# 如果传入的是字符串，则先转换为字典if isinstance(data, str):try:data = json.loads(data)except Exception as e:return f"无法解析天气数据: {e}"# 如果数据中包含错误信息，直接返回错误提示if "error" in data:return f"⚠ {data['error']}"# 提取数据时做容错处理city = data.get("name", "未知")country = data.get("sys", {}).get("country", "未知")temp = data.get("main", {}).get("temp", "N/A")humidity = data.get("main", {}).get("humidity", "N/A")wind_speed = data.get("wind", {}).get("speed", "N/A")# weather 可能为空列表，因此用 [0] 前先提供默认字典weather_list = data.get("weather", [{}])description = weather_list[0].get("description", "未知")return (f"🌍 {city}, {country}\n"f"🌡 温度: {temp}°C\n"f"💧 湿度: {humidity}%\n"f"🌬 风速: {wind_speed} m/s\n"f"⛅ 天气: {description}\n")@mcp.tool()
async def query_weather(city: str) -> str:"""输入指定城市的英文名称，返回今日天气查询结果。:param city: 城市名称（需使用英文）:return: 格式化后的天气信息"""data = await fetch_weather(city)return format_weather(data)@mcp.tool()
async def Add(x:int,y:int)->int:"""执行两个整数的加法运算:param x: 第一个加数:param y: 第二个加数:return: 两数之和"""data=x+yreturn data@mcp.tool()
async def Mul(x:int,y:int)->int:"""执行两个整数的乘法运算:param x: 第一个乘数:param y: 第二个乘数:return: 两数之积"""data=x*yreturn dataif __name__ == "__main__":# 以标准 I/O 方式运行 MCP 服务器mcp.run(transport='stdio')

这里，我们增加了两个自定义的MCP服务工具


@mcp.tool()
async def Add(x:int,y:int)->int:"""执行两个整数的加法运算:param x: 第一个加数:param y: 第二个加数:return: 两数之和"""data=x+yreturn data@mcp.tool()
async def Mul(x:int,y:int)->int:"""执行两个整数的乘法运算:param x: 第一个乘数:param y: 第二个乘数:return: 两数之积"""data=x*yreturn data

然后重新启动

uv run client-mcp.py server.py

以下是相关运行效果

server.py
Processing request of type ListToolsRequest已连接到服务器，支持以下工具: ['query_weather', 'Add', 'Mul']🤖 MCP 客户端已启动！输入 'quit' 退出你: 介绍一下你当前的功能
Processing request of type ListToolsRequest
[{'type': 'function', 'function': {'name': 'query_weather', 'description': '\n    输入指定城市的英文名称，返回今日天气查询结果。\n    :param city: 城市名称（需使用英文）\n    :return: 格式化后的天气信息\n    ', 'input_schema': {'properties': {'city': {'title': 'City', 'type': 'string'}}, 'required': ['city'], 'title': 'query_weatherArguments', 'type': 'object'}}}, {'type': 'function', 'function': {'name': 'Add', 'description': '\n    执行两个整数的加法运算\n    :param x: 第一个加数\n    :param y: 第二个加数\n    :return: 两数之和\n    ', 'input_schema': {'properties': {'x': {'title': 'X', 'type': 'integer'}, 'y': {'title': 'Y', 'type': 'integer'}}, 'required': ['x', 'y'], 'title': 'AddArguments', 'type': 'object'}}}, {'type': 'function', 'function': {'name': 'Mul', 'description': '\n    执行两个整数的乘法运算\n    :param x: 第一个乘数\n    :param y: 第二个乘数\n    :return: 两数之积\n    ', 'input_schema': {'properties': {'x': {'title': 'X', 'type': 'integer'}, 'y': {'title': 'Y', 'type': 'integer'}}, 'required': ['x', 'y'], 'title': 'MulArguments', 'type': 'object'}}}]🤖 OpenAI: <think>
好的，用户问我介绍一下当前的功能。首先，我需要回顾一下提供的工具有哪些。有三个功能：查询天气、加法运算和乘法运算。接下来，我要把这些功能用简单易懂的话解释清楚。用户可能不太清楚每个功能的具体用途，所以得分别说明。比如，查询天气需要输入城市名称，然后返回天气信息。加法和乘法则直接处理两个整数的运算。要注意用词要友好，避免技术术语，让普通用户也能明白。另外，用户可能想知道是否需要特定的参数，比如天气功能需要英文城市名，这点要提到。同时，要提醒用户如果需要帮助，可以进一步询问。最后，保持回答简洁，不需要太多细节，但要覆盖所有功能。
</think>目前我具备以下功能：
1. 天气查询：可输入英文城市名获取当日天气信息
2. 数字运算：支持整数加法（Add）和乘法（Mul）操作
3. 基本交互：能回答简单问题并提供帮助需要我为您执行哪个功能？例如查询北京天气，或进行5+3的计算。你: 计算一下2和3的和是多少
Processing request of type ListToolsRequest
[{'type': 'function', 'function': {'name': 'query_weather', 'description': '\n    输入指定城市的英文名称，返回今日天气查询结果。\n    :param city: 城市名称（需使用英文）\n    :return: 格式化后的天气信息\n    ', 'input_schema': {'properties': {'city': {'title': 'City', 'type': 'string'}}, 'required': ['city'], 'title': 'query_weatherArguments', 'type': 'object'}}}, {'type': 'function', 'function': {'name': 'Add', 'description': '\n    执行两个整数的加法运算\n    :param x: 第一个加数\n    :param y: 第二个加数\n    :return: 两数之和\n    ', 'input_schema': {'properties': {'x': {'title': 'X', 'type': 'integer'}, 'y': {'title': 'Y', 'type': 'integer'}}, 'required': ['x', 'y'], 'title': 'AddArguments', 'type': 'object'}}}, {'type': 'function', 'function': {'name': 'Mul', 'description': '\n    执行两个整数的乘法运算\n    :param x: 第一个乘数\n    :param y: 第二个乘数\n    :return: 两数之积\n    ', 'input_schema': {'properties': {'x': {'title': 'X', 'type': 'integer'}, 'y': {'title': 'Y', 'type': 'integer'}}, 'required': ['x', 'y'], 'title': 'MulArguments', 'type': 'object'}}}]
Processing request of type CallToolRequest[Calling tool Add with args {'x': 2, 'y': 3}]🤖 OpenAI: <think>
好的，用户让我计算2和3的和，我需要确认他们是否真的需要简单的加法。首先，我应该检查是否有其他可能的解释，比如是否有隐藏的数学概念或者陷阱。不过根据常见的数学问题，2加3的和显然是5。用户可能只是想确认基本的加法操作，或者测试我的反应速度。也有可能他们想看看我是否能正确识别简单的算术题。需要确保回答准确且清晰，避免任何混淆。此外，用户可能希望得到即时的反馈，所以直接给出答案并说明过程有助于他们理解。最后，保持回答简洁，符合用户的需求。
</think>2和3的和是5。

可以看到，编写MCP服务端代码其实很简单，如下所示

@mcp.tool()
async def Add(x:int,y:int)->int:"""执行两个整数的加法运算:param x: 第一个加数:param y: 第二个加数:return: 两数之和"""data=x+yreturn data

@mcp.tool()：声明一个装饰器，表明Mul函数是一个工具函数。在某些框架（如 LangChain）中，此类装饰器用于将函数注册为可被智能代理调用的工具，使函数能在自动化流程中被动态调用。

工具说明：工具的注释说明必须要有，这是模型识别并调用工具的关键

"""
执行两个整数的乘法运算
:param x: 第一个乘数
:param y: 第二个乘数
:return: 两数之积
"""

工具函数的具体实现

[C++] traits机制

领域驱动设计(DDD)【2】之项目启动与DDD基本开发流程

AtCoder AT_abc411_c [ABC411C] Black Intervals

多头注意力机制中全连接函数

阿里云ACP认证-数据仓库

如何优化HarmonyOS 5的分布式通信性能？

day44-硬件学习之arm启动代码

3D可视化数字孪生智能服务平台-物联网智控节能控、管、维一体化技术架构

Shell编程中的Ansible常用模块

Rust 和 R 语言的十大应用领域

springboot口腔管理平台

2025.6.21笔记

`provide` 和 `inject` 组件通讯：实现跨组件层级通讯

成长笔记——多串口发送与接收

企业公用电脑登录安全管控的终极方案：ASP操作系统安全登录管控方案

编程基础：耦合

JVM(8)——详解分代收集算法

无线Debugger攻防全解：原理剖析与突破之道

个人博客使用NextWatermark WordPress插件为网站图片自动批量添加水印，保护图片版权

【软考高级系统架构论文】论云上自动化运维及其应用

武汉便宜做网站hlbzx/长春seo

wdcp 网站迁移/百度网页游戏大厅

学生做网站赚钱/今天的新闻联播

flash代码做网站教程/关于友情链接说法正确的是

网站开发费会计科目/热搜排行榜今日排名

网站建设展示型是什么/北京优化网站方法