学习一个新技术，最快的方法还是把他跑起来，看看效果，然后去研究里面的细节，然后再结合官网或者网络上的一些教程做更深入理解。如果时间充足，再把自己学习到的知识，讲给别人听，加深自己的理解，这也是我学习更新的动力所在。
注：基本学习资料是从b-站-赋范空间来的，加上了自己手动测试和一些理解
uv环境准备我们显示直接使用uv来操作，安装uv环境
pip install uv打开cmd，创建并初始化项目目录
uv init mcp-demo使用vscode打开文件目录
mcp项目目录
申请天气查询的apikey访问地址：https://openweathermap.org/
注册登录之后，设置api key
环境变量配置大模型创建.env文件，文件中的配置可以直接使用os.getenv("OPENAI_API_KEY")读取，很方便
设置自己的大模型地址
# 阿里云百炼的模型# BASE_URL="https://dashscope.aliyuncs.com/compatible-mode/v1"# MODEL=qwen-plus# OPENAI_API_KEY="sxxxxxxxxxxxxxx"# 本地ollama部署模型BASE_URL=http://localhost:11434/v1/MODEL=gemma3:27bOPENAI_API_KEY=ollama创建mcp clientimport asyncioimport osimport jsonfrom typing import Optionalfrom contextlib import AsyncExitStackfrom openai import OpenAI from dotenv import load_dotenvfrom mcp import ClientSession, StdioServerParametersfrom 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") # 读取 model if 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 服务器并建立通信，和server.py通信，拿到@mcp.tool()注解的信息 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 服务器上的工具。这里会拿到mcp server所有的工具 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: """ 使用大模型处理查询并调用可用的 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.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 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(): 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())创建mcp serverimport jsonimport httpxfrom typing import Anyfrom mcp.server.fastmcp import FastMCP# 初始化 MCP 服务器mcp = FastMCP("WeatherServer")# OpenWeather API 配置OPENWEATHER_API_BASE = "https://api.openweathermap.org/data/2.5/weather"API_KEY = "xxxxxxxxxxxx" # OpenWeather的API KeyUSER_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__": mcp.run(transport='stdio')说明：一个server可以有多个工具，只需要在方法上加上@mcp.tool()注解即可
安装项目依赖并运行激活虚拟环境
# 创建虚拟环境uv venv# 激活虚拟环境.venv/bin/activate此时项目结构如下
命令行前面会显示环境名
安装需要的包
uv add mcp openai python-dotenv httpx
运行代码，服务端要作为客户端子服务启动
uv run client.py server.py我们查询武汉天气试试
结尾思考根据运行结果，我们会发现这个和function call好像是一样的，的确是这样，但是mcp把各种function call格式化了，以前使用不同的call需要对接不同的文档，接口。现在使用同一种协议，开发效率不仅有提升，而且因为使用mcp协议，很多开源的mcp service我们也能够很快接入。
开发一个强大的智能体直接使用开源的mcp service，想想就很兴奋。
