MCP客户端与服务端初使用——让deepseek调用查询天气的mcp来查询天气

Ne0inhk

— 13 min read
MCP客户端与服务端初使用——让deepseek调用查询天气的mcp来查询天气

本系列主要通过调用天气的mcp server查询天气这个例子来学习什么是mcp，以及怎么设计mcp。话不多说，我们开始吧。主要参考的是B站的老哥做的一个教程，我把链接放到这里，大家如果有什么不懂的也可以去看一下。
https://www.bilibili.com/video/BV1NLXCYTEbj?spm_id_from=333.788.videopod.episodes&vd_source=32148098d54c83926572ec0bab6a3b1d

https://blog.ZEEKLOG.net/fufan_LLM/article/details/146377471

最终的效果:让deepseek-v3使用天气查询的工具来查询指定地方的天气情况

在这里插入图片描述

技术介绍

MCP，即Model Context Protocol（模型上下文协议），是由Claude的母公司Anthropic在2024年底推出的一项创新技术协议。在它刚问世时，并未引起太多关注，反响较为平淡。然而，随着今年智能体Agent领域的迅猛发展，MCP逐渐进入大众视野并受到广泛关注。今年2月，Cursor宣布正式支持MCP功能，这无疑为MCP的推广按下了加速键，使其迅速走进了众多开发人员的视野。
从本质上讲，MCP是一种在智能体Agent开发过程中被广泛认可并遵循的规范。它就如同古代秦始皇推行的“书同文、车同轨”政策一样，通过建立统一的标准和规范，能够极大地提升各方在协作过程中的效率。当开发者们都遵循这一协议进行智能体Agent的开发时，沟通成本大幅降低，开发流程更加顺畅，进而显著提高了智能体Agent的整体开发效率。
截至目前，已经有上千种基于MCP的工具应运而生。在如此强大且不断壮大的MCP生态系统推动下，未来有望迎来一个全新的时代，即人人能够便捷地手工打造Manus的时代。
MCP解决的问题,就是让Agent开发中调用工具更加方便.因为Agent的关键,就是希望它除了聊天之外,能够使用外部工具.之前的function calling的思路,就是创建一个外部的function作为中介,一边传递大模型的请求,一边调用外部工具,最终让大模型能够间接的调用外部工具.

在这里插入图片描述


在这里插入图片描述


这个的思路很简单就不多介绍了,主要是function calling有什么问题?function calling最大的问题就在于它的工具量非常大,一个简单的外部函数往往就要上百行的代码,并且,为了让大模型认识这些外部函数,还需要为每个外部函数编写一个json schema格式的功能说明,此外还要精心设计一个提示词模板才能提高function calling响应的准确率.
所以MCP的目标,就是为了简化这一过程.先统一名称.把大模型的运行环境成为MCP Clinet,也就是MCP客户端,同时,把外部函数运行环境作为MCP Server,MCP服务端.

在这里插入图片描述


然后,统一客户端和服务端的运行规范.并且要求MCP客户端和服务端之间,也统一按照某个既定的提示词模板进行通信.

这样做目前最大的好处就在于MCP服务器的开发成为一个特定的模板,并且只要开发了一个MCP server之后,大家也就都能够直接进行使用.比如查询天气,网页爬取,查询本地SQL等等通用需求.
于是github中的MCP成堆出现,大模型能够使用的工具不断增多,

在这里插入图片描述


b站的九天老师这块讲的非常清楚,关于agent的入门学习理论是很合适的:
https://www.bilibili.com/video/BV1CcBJYtEne/?vd_source=32148098d54c83926572ec0bab6a3b1d

在这里插入图片描述


这里就不再赘述了,主要包含的四个模块及其作用方面.

在这里插入图片描述

环境构建

我看大模型使用的最多的环境是uv的虚拟环境,这里稍微讲一下他们之间的关系.conda是一个跨语言的包和环境的管理工具(不止有python环境),uv则是一个超快的,主要用于python生态,和pip poetry是一类的

在这里插入图片描述


这里我们选择uv来构建,也借着学习一下uv的安装和命令,首先是通过pip来安装uv,python的版本的话优先考虑py3.11的.

pip install uv

安装依赖库

uv pip install requests # 与pip install 类似,但是更快

创建虚拟环境.到这里才是真正使用uv创建虚拟环境:

uv venv myenv

这里不像conda,可以指定自己的python版本,这里默认就是使用当前python版本进行虚拟的.类似于运行了:python -m venv myenv.
然后就可以激活环境:

source myenv/bin/activate

windows中则是:

myenv\Scripts\activate # Windows

然后在虚拟环境中都要通过uv开头的命令来运行文件,这样能够更加高效.

上面主要是uv的一些基础内容,接下来我们正式建立mcp client和mcp server
创建mcp客户端:
我用的是自己在win上安装的py3.8

uv init mcp-client cd mcp-client
在这里插入图片描述


然后创建虚拟环境并进入:

uv venv .venv\Scripts\activate
在这里插入图片描述


接下来安装一些依赖库:

uv add mcp openai python-dotenv

报错了:

在这里插入图片描述

因为mcp 包要求 Python 版本大于等于 3.10，这就导致依赖解析失败。所以我们推荐的还是py3.11的版本.
那怎么通过uv创建不同的python环境呢?
uv没有这样的功能.所以说我们还是需要一个conda,创建一个py3.11的版本,然后重新直线上述过程.还挺无语的,但是要想要uv的快速原型功能也只好这样了…

conda create -n mcp_server python=3.11 conda activate mcp_server

然后重新uv:

uv init mcp-client cd mcp-client uv venv .venv\Scripts\activate uv add mcp openai python-dotenv
在这里插入图片描述


python双重环境…
安装完毕之后:

在这里插入图片描述


创建客户端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 Key self.base_url = os.getenv("BASE_URL")# 读取 BASE YRL self.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 API response = await asyncio.get_event_loop().run_in_executor( None, lambda: self.client.chat.completions.create(model=self.model, messages=messages, stream=False ))return response.choices[0].message.content except 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 response = await self.process_query(query)# 发送用户输入到 OpenAI API 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())

然后在文件夹中新建.env文件,写入对应的api-key:
我测试过了,deepseek官网的api才支持后续的工具的使用,或者是昂贵的openai和claude,这里就不考虑了

BASE_URL=https://api.deepseek.com MODEL=deepseek-chat OPENAI_API_KEY="sk-xxx"

运行:

uv run client.py
在这里插入图片描述


然后是服务端的核心部分,我们以天气调用的工具为例:
首先你需要一个openweather的API来查询天气:https://openweathermap.org/api

在这里插入图片描述


注册一个账号之后去生成一个API即可

在这里插入图片描述

服务端的依赖项安装:

uv add mcp httpx

新建一个server.py:

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 ="xxx"# 请替换为你自己的 OpenWeather API Key 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_new.py:

import asyncio import os import json 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 Key self.base_url = os.getenv("BASE_URL")# 读取 BASE YRL self.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)# 创建OpenAI client self.session: Optional[ClientSession]= None self.exit_stack = AsyncExitStack() async def connect_to_server(self, server_script_path: str): """连接到 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_transport self.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.tools print("\n已连接到服务器，支持以下工具:", [tool.name fortoolin tools]) async def process_query(self, query: str) -> str: """ 使用大模型处理查询并调用可用的 MCP 工具 (Function Calling)""" print(11111) messages =[{"role":"user", "content": query}] print(2222) response = await self.session.list_tools() print(33333) available_tools =[{"type":"function", "function":{"name": tool.name, "description": tool.description, "input_schema": tool.inputSchema }}fortoolin response.tools] print(available_tools) print(44) response = self.client.chat.completions.create(model=self.model, messages=messages, tools=available_tools ) print(5555)# 处理返回的内容 content = response.choices[0]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)# 执行工具 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 return content.message.content # async def process_query(self, query: str) -> str:# """调用 OpenAI API 处理用户查询"""# messages = [{"role": "system", "content": "你是一个智能助手，帮助用户回答问题。"},# {"role": "user", "content": query}]# try:# # 调用 OpenAI API# response = 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.content# except 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 print("chat ing") response = await self.process_query(query)# 发送用户输入到 OpenAI API 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(): 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 sys asyncio.run(main())

要么新启动一个终端来运行server部分,需要重新进入虚拟环境.当然,uv也支持同时运行,所以可以直接运行:

uv run client_new.py server.py
在这里插入图片描述


最终效果:

在这里插入图片描述

Read more

Transformer实战（9）——Transformer分词算法详解

Transformer实战（9）——Transformer分词算法详解

Transformer实战（9）——Transformer分词算法详解 * 0. 前言 * 1. 子词分词算法 * 2. 加载预训练分词器 * 3. 常见字词分词算法 * 3.1 字节对编码 * 3.2 WordPiece * 3.3 SentencePiece * 4. 使用 tokenizers 库训练分词器 * 4.1 训练 BPE * 4.2 训练 WordPiece * 4.3 空分词管道 * 小结 * 系列链接 0. 前言 在自然语言处理领域，高效准确的分词算法是构建强大语言模型的基础。随着 Transformer 架构的广泛应用，子词分词算法已成为处理多语言文本和稀有词汇的关键技术。本文将从理论到实践，全面解析现代自然语言处理 (Natural Language Processing,

By Ne0inhk

Qwen3-Reranker-0.6B实战手册：Gradio Web服务+Python API双模式调用详解

Qwen3-Reranker-0.6B实战手册：Gradio Web服务+Python API双模式调用详解 1. 快速了解Qwen3-Reranker-0.6B Qwen3-Reranker-0.6B是通义千问团队最新推出的文本重排序模型，专门用于提升搜索和检索任务的效果。这个模型虽然只有6亿参数，但在文本重排序任务上表现出色，能够智能地判断文档与查询的相关性，并重新排序输出最相关的结果。 简单来说，它就像一个智能的文档筛选助手。你给它一个问题（比如"北京有什么好玩的地方？"）和一堆候选答案，它能帮你找出最相关的几个答案，并按相关性从高到低排列。 这个模型支持100多种语言，能处理长达32K字符的文本，而且部署简单，只需要1.2GB的存储空间。无论你是做中文搜索、英文检索，还是处理代码片段，它都能胜任。 2. 环境准备与快速部署 2.1 系统要求 在开始之前，确保你的环境满足以下基本要求： * Python 3.8或更高版本（推荐Python 3.10） * 至少4GB内存（处理大量文档时需要更多）

By Ne0inhk

【性能提升300%】：基于Python的多模态数据分块存储优化实践

第一章：【性能提升300%】：基于Python的多模态数据分块存储优化实践 在处理图像、文本与传感器数据等多模态数据时，传统单一存储结构常导致I/O瓶颈和内存浪费。通过引入分块存储策略与异步写入机制，结合Python中的HDF5与Zarr库，可显著提升数据读写效率。 设计高效的分块策略 合理的分块大小能平衡磁盘访问与内存占用。对于高分辨率图像与长序列文本，采用动态分块策略： * 图像数据：按 (64, 64, 3) 分块，适配常见卷积网络输入 * 文本序列：按 token 数量切分为长度为512的块 * 传感器数据：按时间窗口（如每秒100条）进行分段 使用Zarr实现压缩与并行访问 Zarr支持分层数组存储，并内置多种压缩算法。以下代码展示如何将多模态数据写入Zarr容器： # 导入必要库 import zarr import numpy as np # 创建根组 root = zarr.group() # 存储图像块，使用Blosc压缩 images = root.zeros('

By Ne0inhk
基于Python的医院运营数据可视化平台：设计、实现与应用（上）

基于Python的医院运营数据可视化平台：设计、实现与应用（上）

一、引言 1.1 研究目的与意义 在信息技术日新月异的当下，医疗行业正处于深刻的变革之中，逐渐朝着信息化、智能化方向大步迈进。医院每天都会产生海量的数据，涵盖患者信息、诊疗记录、药品库存、设备使用状况等多个关键领域。这些数据宛如一座蕴藏丰富的宝藏，若能加以科学有效的管理与分析，将为医院的运营管理提供强大的支持，成为提升医疗服务质量、优化资源配置的关键要素。然而，传统的数据处理方式，如过度依赖 Excel 表格和简单的统计工具，在面对如此庞大且复杂的数据时，显得力不从心，效率极为低下，难以满足医院日益增长的复杂分析需求。 基于此，本研究旨在构建一个基于 Python 的医院运营数据可视化平台，将 Python 语言的强大功能与数据可视化技术深度融合，为医院运营管理开辟全新的路径。Python 作为一种高级编程语言，以其简洁性、易读性以及丰富多样的扩展库，如 Pandas、Matplotlib 等，在数据处理和可视化领域展现出卓越的优势，能够高效地实现数据清洗、分析以及可视化图表的生成。 本研究具有重要的现实意义。对于医院管理而言，该平台能够极大地提高运营管理的效率。

By Ne0inhk