从零到一MCP快速入门实战【1】

参考视频

• MCP入门到实战

一 MCP入门介绍

• MCP（Model Context Protocol）模型上下文协议，由Claude母公司Anthropic于2024年11月正式提出。
• 本质上来说，MCP是一种技术协议，一种智能体Agent开发过程中共同约定的一种规范。在统一的规范下，大家的协作效率大幅提高，最终提升智能体Agent的开发效率。截止目前，已上千种MCP工具诞生，在强悍的MCP生态加持下，人人手搓Manus的时代即将到来。
• MCP解决了Agent开发中调用外部工具技术门槛过高的问题。调用外部工具是大模型进化为Agent的关键，如果不能使用外部工具，大模型只能是简单的聊天机器人。由于底层技术限制，大模型本身无法和外部工具进行通信，因此Function Calling就是创建一个外部函数作为中介，一边传递大模型的请求，另一边调用外部工具，最终让大模型能够间接调用外部工具。
  
• 例如，当要查询当前天气时，让大模型调用外部工具的function calling的过程，如图所示：

• Function calling存在的问题是编写外部工具工作量太大，一个简单的外部函数往往需要上百行代码，而且，为了让大模型认识外部函数，还需要额外为每个函数编写一个JSON Schema格式的功能说明，此外，为提高Function calling响应准确率，还需要精心设计一个提示词模板。而MCP的目标就是在Agent开发过程中，让大模型更加边界调用外部工具。为此，MCP提出了两个方案——统一Function calling的运行规范+。

• 首先是名称统一，MCP把大模型运行环境称作MCP Client（MCP客户端），把外部函数环境称为MCP Server(MCP服务器)。然后，统一MCP客户端和服务器的运行规范，并且要求MCP客户端和服务器中间，也统一按照某个既定的提示词模板进行通信。
  

• 为进一普及MCP协议，Anthropic还提供一整套MCP客户端、服务器开发的SDK，并且支持Python、TS和Java等多种语言，借助SDK，仅需几行代码，就可以快速开发一个MCP服务器。

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

二 Function Calling执行流程

三 Agent开发技术体系

Agent（主体）与四个模块（Tools模块、Planing模块、Memory模块、Action模块）之间的关系，其中：

• Tools模块 负责让大模型可以连接外部工具。（使得大模型能够连接数据库；使得大模型编写的SQL语句能够直接运行；）
• Memory模块 负责管理大模型对话时的记忆。（管理多轮对话，记住此前对话内容；负责关联本地知识库，理解行业“黑话”；控制模型对话输出语气风格、文本长短等）
• Planing模块 负责管理大模型行动的基本流程。（判断复杂任务如何拆解；判断何时该进行外部知识库检索；判断何时该编写SQL代码；代码出错时应该如何反馈执行）
• Action模块 负责规划大模型的行动。（搭建WorkFlow、即执行流程；明确多个Agent之间协作关系；简单情况下需要搭建链式结构；复杂情况下需要搭建图结构）

---

四 MCP客户端开发流程

4.1 UV基本知识

4.1.1 UV工具入门使用

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

---

UV的特点：

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

---

4.1.2 UV安装流程

方法1：使用pip安装

pip install uv

方法2：使用curl安装(自动下载uv并安装到/usr/local/bin)

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

4.1.3 UV的基本用法

• 安装Python依赖

uv pip install requests

• 创建虚拟环境

uv venv myenv

• 激活虚拟环境

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

• 直接运行Python项目

uv run python script.py

4.2 MCP极简客户端搭建流程

4.2.1 创建MCP客户端项目

# 创建项目目录
uv init mcp-client
cd mcp-client

4.2.2 创建MCP客户端虚拟环境

• 在mcp-client目录下执行

uv venv # 创建虚拟环境
source .venv/bin/activate # 1 linux/macos 激活虚拟环境
.venv\Scripts\activate # 2 windows激活虚拟环境

• 通过add方法在虚拟环境中安装相关库

# 安装 MCP SDK
(mcp-client) D:\Code\mcp-study\mcp-client>uv add mcp

4.2.3 编写基础MCP客户端

• 在当前项目主目录中创建client.py

import asyncio
from mcp import ClientSession
from contextlib import AsyncExitStackclass MCPClient:def __init__(self):"""初始化MCP客户端"""self.session = Noneself.exit_stack = AsyncExitStack()async def connect_to_mock_server(self):"""连接MCP服务器的连接"""print("MCP客户端初已始化，但未连接到服务器")async def chat_loop(self):"""运行交互式聊天循环"""print("\nMCP客户端已启动！输入'quit'退出")while True:try:query=input("\nQuery:").strip()if query.lower() == "quit":breakprint(f"Mock Response: your question:{query}")except Exception as e:print(f"Error: {e}")async def cleanup(self):"""清理资源"""await self.exit_stack.aclose()async def main():client = MCPClient()try:await client.connect_to_mock_server()await client.chat_loop()finally:await client.cleanup()if __name__ == "__main__":asyncio.run(main())

• 运行极简的MCP客户端

(mcp-client) D:\Code\mcp-study\mcp-client>uv run client.py

五 MCP客户端接入大模型流程

• 尝试在客户端中接入OpenAl和DeepSeek等在线模型进行对话。需要注意的是，由于OpenAl和DeepSeek调用方法几乎完全一样，因此这套服务器clilent代码可以同时适用于GPT模型和DeepSeek。

5.1 新增依赖

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

(mcp-client) D:\Code\mcp-study\mcp-client>uv add mcp openai python-dotenv

5.2 创建.env文件

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

BASE_URL="https://api.siliconflow.cn/v1/chat/completions"
MODEL=deepseek-ai/DeepSeek-V3
OPENAI_API_KEY="sk-xxx"

5.3 初步连接大模型

• 修改客户端代码如下：

import asyncio
import os
from openai import OpenAI
from dotenv import load_dotenv
from contextlib import AsyncExitStack# 加载env文件，确保配置正确
load_dotenv()class MCPClient:def __init__(self):"""初始化MCP客户端"""self.exit_stack = AsyncExitStack()self.base_url=os.getenv("BASE_URL")self.model = os.getenv("MODEL")self.openai_api_key = os.getenv("OPENAI_API_KEY")if not self.openai_api_key:raise ValueError("OPENAI_API_KEY未设置")self.client=OpenAI(api_key=self.openai_api_key, base_url=self.base_url)async def connect_to_mock_server(self):"""连接MCP服务器的连接"""print("MCP客户端初已始化，但未连接到服务器")async def process_query(self, query:str)->str:"""调用OpenAI API处理用户查询"""message=[{"role": "system","content": "你是一个智能助手，默认使用中文回答问题."},{"role": "user","content": query,}]try:response=await asyncio.get_event_loop().run_in_executor(None,lambda: self.client.chat.completions.create(model=self.model,messages=message))return response.choices[0].message.contentexcept Exception as e:print(f"调用OpenAI API时出错: {e}")async def chat_loop(self):"""运行交互式聊天循环"""print("\nMCP客户端已启动！输入'quit'退出")while True:try:query=input("Query:").strip()if query.lower() == "quit":breakresponse=await self.process_query(query)print(f"\nResponse: {response}\n")except Exception as e:print(f"Error: {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())

MCP客户端已启动！输入'quit'退出
Query:计算3.33*3=
Response: 要计算 \( 3.33 \times 3 \)，可以按照以下步骤进行：
\[
\begin{align*}
3.33 \times 3 &= (3 + 0.33) \times 3 \\
&= 3 \times 3 + 0.33 \times 3 \\
&= 9 + 0.99 \\
&= 9.99
\end{align*}
\]
因此，最终答案为：
\[
\boxed{9.99}
\]

六 MCP客户端接入ollama、vLLM模型

• 将ollama、vLLM等模型调度框架接入MCP的client。由于ollama和vLLM均支持OpenAlAPI风格调用方法，因此上述client.py并不需要进行任何修改，我们只需要启动响应的调度框架服务，然后修改.env文件即可。

1. 启动ollama，如下ollama已经运行：

   ollama start
   Error: listen tcp 127.0.0.1:11434: bind: Only one usage of each socket address (protocol/network address/port) is normally permitted.
   

2. 修改.env文件

   BASE_URL="http://localhost:11434/v1/"
   MODEL="deepseek-r1:14b"
   OPENAI_API_KEY="ollama"
   

3. 运行client

(mcp-client) D:\Code\mcp-study\mcp-client>uv run client.py

---

• 接入vLLM模型配置文件内容，其他操作和上面的相同。

  BASE_URL="http://localhost:8000/v1"
  MODEL="deepseek-r1:14b"
  OPENAI_API_KEY=EMPTY
  

七 MCP天气查询服务器server与使用

7.1 MCP服务器

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

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

7.2 MCP服务器通讯机制

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

• 通信方式

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

• 连接特性：

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

• 适用场景：

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

• SSE仅支持服务器向客户端的单向通信，而WebSocket则支持双向通信。

传输方式	是否需要同时启动服务器	是否支持远程连接	适用场景
stdio(标准输入输出)	需要	不支持	本地通信，延迟低，告诉交互
http(网络API)	不需要	支持	分布式架构，远程通信

• stdio模式提供了一种简单、高效的本地通信方式，适用于客户端和服务器在同一环境下运行的情况。而对于分布式或远程部署的场景，基于HTTP和SSE的传输方式则更为合适。

7.3 天气查询服务器Server创建流程

• 创建一个天气查询的服务器，通过openweather，创建一个能够实时查询天气的服务器（server），并适用stdio方式进行通信。

  curl -s "https://api.openweathermap.org/data/2.5/weather？q=Beijing&units=metric&appid=055bcd72c5dd30682f75465070ae0f46"
  

• 执行结果：

  {"coord": {"lon": 116.3972,"lat": 39.9075},"weather": [{"id": 804,"main": "Clouds","description": "阴，多云","icon": "04n"}],"base": "stations","main": {"temp": 22.36,"feels_like": 22.77,"temp_min": 22.36,"temp_max": 22.36,"pressure": 1007,"humidity": 81,"sea_level": 1007,"grnd_level": 1002},"visibility": 10000,"wind": {"speed": 1.42,"deg": 26,"gust": 3.23},"clouds": {"all": 100},"dt": 1753014180,"sys": {"country": "CN","sunrise": 1752958921,"sunset": 1753011546},"timezone": 28800,"id": 1816670,"name": "Beijing","cod": 200
  }
  

7.3.1 服务器依赖安装和代码编写

1. 创建项目

   uv mcp-weather
   cd mcp-weather
   .venv\Scripts\activate
   

2. 在当前虚拟环境中添加如下依赖：

   uv add mcp httpx openai python-dotenv pypinyin
   

• MCP基本执行流程：
  

3. 创建server服务器代码文件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 ="055bcd72c5dd30682f75465070ae0f46"
   USER_AGENT = "weather-app/1.0"async def fetch_weather(city: str) -> dict[str, Any]|None:"""获取天气信息"""params={"q": city,"appid": API_KEY,"units": "metric","lang": "zh_cn"}headers={"User-Agent": USER_AGENT}async with httpx.AsyncClient() as client:response = await client.get(OPENWEATHER_API_BASE, params=params, headers=headers,timeout=1000)if response.status_code == 200:return response.json()else:return Nonedef format_weather(data: dict[str,Any] | str)->str:"""解析天气数据字典，提取关键信息并格式化输出。功能：对可能缺失的嵌套数据字段进行容错处理，确保返回内容完整。参数：data: 天气API返回的原始数据字典返回：格式化后的天气信息字符串"""# 基础位置信息（城市、国家）- 缺失时显示"未知"city = data.get("name", "未知")  # 城市名称（顶层字段）country = data.get("sys", {}).get("country", "未知")  # 国家代码（嵌套在sys字段中）# 天气核心指标 - 缺失时显示"N/A"（Not Available）main_data = data.get("main", {})  # 提取main字段（包含温度、湿度等）temperature = main_data.get("temp", "N/A")  # 温度humidity = main_data.get("humidity", "N/A")  # 湿度wind_data = data.get("wind", {})  # 提取wind字段（包含风速等）wind_speed = wind_data.get("speed", "N/A")  # 风速# 天气描述 - weather字段可能为空列表，默认返回第一个元素的描述weather_list = data.get("weather", [{}])  # 提取weather数组（默认空字典避免索引错误）weather_description = weather_list[0].get("description", "未知")  # 天气状况描述# 格式化输出字符串（使用f-string拼接，添加emoji直观展示）weather_info = (f"🌍 {city}, {country}\n"f"🌡️ 温度：{temperature}℃\n"f"💧 湿度：{humidity}%\n"f"💨 风速：{wind_speed} m/s\n"f"☁️ 天气：{weather_description}\n")return [e-mail]()
   async def query_weather(city: str) -> str:"""查询天气信息并返回结果"""weather_data = await fetch_weather(city)if weather_data:return format_weather(weather_data)else:return "无法获取天气信息。请检查城市名称是否正确。"if __name__=="__main__":mcp.run(transport='stdio')
   

7.3.2 客户端代码编写

import asyncio
import os
import json
import sys
from typing import Optional
from  contextlib import AsyncExitStack
from openai.types.chat import ChatCompletionToolParam
from openai import OpenAI
from dotenv import load_dotenvfrom mcp import ClientSession,StdioServerParameters
from mcp.client.stdio import stdio_clientfrom pypinyin import lazy_pinyin, Style# 加载env文件，确保配置正确
load_dotenv()class MCPClient:def __init__(self):"""初始化MCP客户端"""self.write = Noneself.stdio = Noneself.exit_stack = AsyncExitStack()self.base_url=os.getenv("BASE_URL")self.model = os.getenv("MODEL")self.openai_api_key = os.getenv("OPENAI_API_KEY")if not self.openai_api_key:raise ValueError("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):"""连接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 = [ChatCompletionToolParam(type="function",function={"name": tool.name,"description": tool.description,"parameters": tool.inputSchema})for tool in response.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)# 如果调用的是 query_weather 工具，处理城市名称if tool_name == "query_weather" and "city" in tool_args:city = tool_args["city"]# 简单判断是否为中文城市名if any('\u4e00' <= c <= '\u9fff' for c in city):# 转换为拼音，首字母大写pinyin_city = ''.join([word.capitalize() for word in lazy_pinyin(city)])tool_args["city"] = pinyin_city# 执行工具result=await self.session.call_tool(tool_name, tool_args)print(f"\n\n[Calling Tool: {tool_name} with args: {tool_args}]")# 将工具调用和结果添加到消息历史中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.contentasync def chat_loop(self):"""运行交互式聊天循环"""print("\nMCP客户端已启动！输入'quit'退出")while True:try:query=input("\n you:").strip()if query.lower() == "quit":breakresponse=await self.process_query(query)print(f"\n ai: {response}")except Exception as e:print(f"\n Error: {e}")async def cleanup(self):"""清理资源"""await self.exit_stack.aclose()async def main():if len(sys.argv)<2:print("Usage: python client.py <server_address>")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__":asyncio.run(main())

7.4 测试运行

cd ./mcp-weather
source .venv/bin/activate

uv run client.py server.py

(mcp-weather) D:\Code\mcp-study\mcp-weather>uv run client.py server.py已连接到服务器，支持以下工具： ['query_weather']MCP客户端已启动！输入'quit'退出you:请问北京今天天气如何？[Calling Tool: query_weather with args: {'city': 'BeiJing'}]ai: 北京今天的天气情况如下：🌍 **北京，中国**
🌡️ **温度**：22.85℃
💧 **湿度**：74%
💨 **风速**：2.14 m/s
☁️ **天气状况**：阴天，多云请根据实际需要增减衣物，出行注意安全！