前言

之前参加一个智能体大赛但是没有怎么学习过智能体的设计并且时间只有五、六天就设计了一个简单的多智能体系统（如果有感兴趣的同学可以看看gitee仓库：https://gitee.com/dragon-airgitee/open-hot-spot.git）。现在想要好好的了解一下智能体，目前知道MCP技术很强大就想初步了解一下其原理，以下是我通过官方文档以及一些相关博客的学习记录下自己对这个技术的简单了解。

如果文章能帮到其他小伙伴，那是我的荣幸。文中如有不当的地方欢迎大家友好讨论，对于不妥我会积极修改。（有部分图片和实现方案及操作步骤取自其他博客，作者均会在后面引用说明）

大佬们有更好的解决见解和想法，也希望能够指点指点~~~~

一、什么是MCP

MCP 起源于 2024 年 11 月 25 日 Anthropic 发布的文章：Introducing the Model Context Protocol

MCP （Model Context Protocol，模型上下文协议）定义了应用程序和 AI 模型之间交换上下文信息的方式。用来在 AI 大模型和数据源之间建立安全双向的链接。它规范了应用程序向大模型（LLM） 提供上下文的方式。MCP 就像 AI 应用程序的 USB-C 端口一样。正如 USB-C 提供了一种标准化的方式将您的设备连接到各种外围设备和配件一样，MCP 也提供了一种标准化的方式将 AI 模型连接到不同的数据源和工具。MCP 可帮助您在大模型（LLM）之上构建代理和复杂的工作流。

可以看出，MCP 就是以更标准的方式让 LLM Chat 使用不同工具，更简单的可视化如下图所示，这样你应该更容易理解“中间协议层”的概念了。Anthropic 旨在实现 LLM Tool Call 的标准。

二、为什么是MCP

我在一篇知乎中看见一个观点“MCP 的出现是 prompt engineering 发展的产物”。我感觉很有道理，如果我们提供更结构化的上下文信息给模型，他的 performance 提升是显著的。并且我们在构造 prompt 时，也希望能提供一些来自不同功能平台的信息（比如本地文件，数据库，一些网络实时信息等）给模型，这样模型更容易理解我们想要场景中真实问题，并且输出贴近我们想法的答案。

许多 LLM 平台（如 OpenAI、Google）为了克服手工 prompt 的局限性引入了function call 功能。这一机制允许模型在需要时调用预定义的函数来获取数据或执行操作，显著提升了自动化水平。

但是 function call 也有其局限性（

1. 函数定义：定义并实现函数，明确函数名，函数功能描述，参数，参数类型定义

2. 请求大模型并注册函数： 在api调用大模型时，通过tools参数将函数的定义信息传递给大模型

3. 大模型判断与选择： 大模型收到用户请求，并决定是否调用函数，当需要调用函数时，大模型会在回包中返回需要调用的函数名称，函数的参数等信息

4. 执行函数： 本地代码解析大模型回包，当大模型返回需要执行函数的命令时，本地代码按照大模型的函数名，参数，执行函数调用

5. 返回结果： 将函数结果发送给大模型，大模型生成最终的回答

），我认为是 function call 平台依赖性强，不同 LLM 平台的 function call API 实现差异较大。例如，OpenAI 的函数调用方式与 Google 的不兼容，开发者在切换模型时需要重写代码，增加了适配成本。除此之外，还有安全性，交互性等问题。

数据与工具本身是客观存在的，只不过我们希望将数据连接到模型的这个环节可以更智能更统一。Anthropic 基于这样的痛点设计了 MCP，充当 AI 模型的"万能转接头"，让 LLM 能轻松的获取数据或者调用工具。更具体的说 MCP 的优势在于：

• 生态 —— MCP 提供很多现成的插件（punkpeye/awesome-mcp-servers：MCP 服务器的集合。），AI 可以直接使用。
• 统一性 -——不限制于特定的 AI 模型，任何支持 MCP 的模型都可以灵活切换。
• 数据安全 -——你的敏感数据留在自己的电脑上，不必全部上传。（因为我们可以自行设计接口确定传输哪些数据）

三、MCP的架构

官方的架构图：

MCP 采用 C/S 架构，包含 MCP 主机（Hosts）、MCP 客户端（Clients）和 MCP 服务器（Servers） 三个核心组件。

MCP 主机是发起请求的大模型（LLM）应用程序，如 Claude Desktop、IDE 或 AI 工具等；

MCP 客户端在主机程序内部，与 MCP 服务器保持 1:1 的连接；

MCP 服务器则为 MCP 客户端提供上下文、工具和 prompt 信息。

MCP通信基于JSON-RPC 2.0协议，支持请求、响应和通知三种消息类型，确保通信的标准化和一致性。
Local MCP Server（本地数据源）：MCP 服务器可以安全访问本地计算机上的文件、数据库和服务。
Remote MCP Server（远程服务源）：MCP 服务器可以通过互联网（例如通过 API）连接到外部系统。

整个流程是这样的：你的问题 → Host→ AI 大模型 → 需要文件信息 → MCP Client 连接 → 文件系统 MCP Server → 执行操作 → 返回结果 → AI 大模型 → 显示。

四、使用MCP

前面是我在一个月前完成的草稿，第四部分是我为了复盘最近学习到的案例所写的。我参考的是公众号：“大模型真好玩”发布的案例大家可以关注一下我感觉每次看他的推文总有一些收获。

Anthropic 也为我们提供了一个基于 LLM 的 MCP Server 的开发实践：Building MCP with LLMs - Model Context Protocol

我在这里复现的案例是公众号：“大模型真好玩”发布的基于 MCP Http SSE模式的天气助手智能体开发实战。这个案例十分简单和基础感觉很适合我这样的新手操作。

1.环境准备

1. 使用deepseek作为大模型客户端，申请DeepSeek-V3的API Key。 获取当前的天气情况需要调用心知天气的API，申请心知天气API免费私钥。

2. 获取当前的天气情况需要调用心知天气的API，申请心知天气API免费私钥。

   （1）打开心知天气心知天气 - 高精度气象数据 - 天气数据API接口 - 行业气象解决方案官网，注册登录并点击控制台

        （2）在控制台左侧产品管理栏中点击添加产品

       （3）申请免费版的API，点击左侧免费版，就可以看到API私钥

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

（1）执行命令安装uv

pip install uv

（2）执行命令初始化项目

uv init mcp-weather

（3）进行项目文件夹，

cd mcp-weather

（4）执行命令创建虚拟环境

uv venv

（5）执行命令激活虚拟环境

.venv\Scripts\activate

(4) 执行命令安装开发使用的python包

uv add mcp, uv add requests, uv add openai

 2.编写MCP服务端代码

在项目目录下新建server.py文件并编写如下代码：

import json
import aiohttp
from typing import Any
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("WeatherServer")

@mcp.tool()
async def get_weather(city: str):
    """
    输入指定城市的名称，返回当前天气情况
    """
    url = "https://api.seniverse.com/v3/weather/now.json"
    params = {
        "key": "你的心知天气私钥",
        "location": city,
        "language": "zh-Hans",
        "unit": "c"
    }

    async with aiohttp.ClientSession() as session:
        async with session.get(url, params=params) as resp:
            data = await resp.json()
            temperature = data['results'][0]['now']
            return temperature  # FastMCP 会自动序列化
            

if __name__ == "__main__":
    mcp.run(transport="sse")

注意使用http sse模式mcp服务端和使用stdio模式mcp服务端只有启动代码有差别。http sse的启动代码为mcp.run(transport="sse"）, stdio的启动代码为mcp.run(transport="stdio")。

3.编写MCP客户端代码

在项目目录下新建client.py文件, 编写如下代码:

import asyncio
import json
from typing import Optional
from contextlib import AsyncExitStack
from openai import OpenAI

from mcp import ClientSession,StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.client.sse import sse_client

class MCPClient:
    def __init__(self):
        """初始化MCP客户端"""
        self.exit_stack = AsyncExitStack()
        self.openai_api_key = "你的大模型API"# 调用模型的api_key
        self.base_url = "https://api.deepseek.com"
        self.model = "deepseek-chat"
        self.client = OpenAI(api_key=self.openai_api_key,base_url=self.base_url)
        self.session: Optional[ClientSession] = None # Optional提醒用户该属性是可选的，可能为None
        self.exit_stack = AsyncExitStack() # 用来存储和清楚对话中上下文的，提高异步资源利用率

    async def connect_to_sse_server(self, server_url):
        sse_transport = await self.exit_stack.enter_async_context(
            sse_client(server_url)
        )
        self.write,self.read = sse_transport
        self.session = await self.exit_stack.enter_async_context(
            ClientSession(self.write,self.read)
        )
        await self.session.initialize() # 与服务器建立sse连接
        # 列出MCP服务器上的工具
        response = await self.session.list_tools()
        tools = response.tools
        print("\n已连接到服务器，支持以下工具: ",[tool.name for tool in tools])#打印服务端可用的工具


    async def process_query(self,query:str):
        """使用大模型处理查询并调用MCP Server可用的MCP工具"""
        messages = [{"role": "user", "content": query}]
        response = await self.session.list_tools()

        available_tools = [{
            "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.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}]")
            # 将模型返回的调用工具的对话记录保存在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用户: ")
                if query.lower() == "quit":
                    break
                response = await self.process_query(query)
                print(f"\nDeepSeek: {response}")
            except Exception as e:
                print(f"\n发生错误: {e}")

    async def clean(self):
        """清理资源"""
        await self.exit_stack.aclose()

async def main():
    if len(sys.argv) < 2:
        print("使用方法是：python client.py server_url")
        sys.exit(1)

    client = MCPClient()
    try:
        await client.connect_to_sse_server(sys.argv[1])
        await client.chat_loop()
    finally:
        await client.clean()
        
if __name__ == "__main__":
    import sys
    asyncio.run(main())

4.启动服务端和客户端

1.激活虚拟环境的mcp-weather目录下执行uv run server.py命令，可以看到服务端开启了8000端口，在浏览器中输入http://localhost:8000/sse进行测试, 当客户端访问该url会触发服务端处理sse连接，同时服务端也不断的ping维持长连接。

2. 执行命令

uv run client.py http://localhost:8000/sse

第二个参数传入服务端sse的url, 用来建立客户端和服务端的连接，连接成功并返回正确结果，我们基于http sse模式的天气助手智能体就开发完成啦！