应用篇| MCP为智能体插上翅膀

随着OpenManus、owl等通用智能体火爆出圈，mcp协议逐渐被agent框架深度应用。

大靠山

1234人浏览 · 2025-06-27 20:47:08

大靠山 · 2025-06-27 20:47:08 发布

01 AI Agent与MCP关系

随着OpenManus、owl等通用智能体火爆出圈，mcp协议逐渐被agent框架深度应用。

那LLM如何和外部工具整合在一起，之前是利用OpenAI推出的 Function Calling（函数调用），Function Calling 指的是 AI 模型根据上下文自动执行函数的机制。Function Calling 充当了 AI 模型与外部系统之间的桥梁，Function Calling相当于提供了一个跑腿小弟给这个专家：

但不同的模型有不同的 Function Calling 实现，代码集成的方式也不一样。FC是由不同的 AI 模型平台来定义和实现。Function Calling 不一样具体的 API 实现，会出现不通用的问题，导致普及困难且需要重复开发。而且，为了让大模型“认识”这些外部函数，我们还要额外为每个外部函数编写一个JSON Schema格式的功能说明，此外，我们还需要精心设计一个提示词模版，才能提高Function calling响应的准确率。

为解决上述问题，Anthropic于2024年11月推出开源标准Model Context Protocol（MCP），旨在通过标准化方法简化AI与外部数据源的连接。MCP的核心目标是消除多API管理的复杂性，为AI系统提供通用访问接口，无需重复开发定制集成。

02 MCP协议是什么

MCP全称是Model Context Protocol，模型上下文协议，由Claude母公司Anthropic于2024年11月正式提出。旨在标准化AI模型与数据源、工具之间的交互方式，被业界视为“AI时代的USB-C接口”。做过微服务全链路跟踪的开发者应该也接触过上下文协议，这些所谓上下文协议，跟大部分协议如Tcp/IP, http协议等一样的目标，统一规范或者标准。开发者 A 依照MCP规范写出来的 Function ，可以直接开放给开发者 B 使用，相当于把原本独立的 Function 开发过程变成一定意义上的协同开发，支持开发者们把各自为 Agent 实现过程中写的 Function 开源，大幅提升了开发效率，不用再去重复造轮子。按照中国人的历史典故来总结就是“车同轨、书同文”。

MCP协议有什么好处？其实上面内容有涉及一些，下面我们深入总结一下，MCP协议核心价值在于解决传统AI开发中四大痛点：

消除技术栈割裂：通过统一协议替代不同框架（如TensorFlow/PyTorch）的私有API，实现跨框架兼容；

**突破数据孤岛：**允许AI模型安全访问本地文件、数据库及云端服务，构建双向数据通道；

**简化开发流程：**以标准化协议替代重复的Function Calling代码，降低70%以上的集成成本，使得开发者可以快速构建复杂的 AI 流程。这种方式提高了开发效率，降低了学习成本。

**强调数据安全性：**MCP 强调数据安全，支持身份验证和权限管理，确保敏感数据不会被未经授权的访问。

mcp主要干了以下2件事：

统一命名：将大模型的运行环境称作做MCP Client，Function叫做MCP Server；

统一开发规范: 将大模型与Function之间的交互规范统一为一个范式。即要求MCP客户端和服务器之间，统一按照某个既定的提示词模板进行通信。

通过以上2种方式统一规范，可以避免MCP服务器的重复开发，其实本质是避免外部函数重复编写。例如，像查询天气、网页爬取、查询本地MySQL数据库这种通用的需求，大家有一个人开发了一个服务器就好，开发完大家都能复制到自己的项目里来使用，不用每个人每次都单独写一套。这可是促进全球AI开发者共同协作的好事儿，很快，GitHub上就出现了海量的已经开发好的MCP 服务器，从SQL数据库检索、到网页浏览信息爬取，从命令行操作电脑、到数据分析机器学习建模等。

03 MCP开源实现

除了MCP****协议本身：作为开源协议，MCP的代码和规范已公开，任何人都可以自由使用、修改和扩展。有官方开源以及社区开源的实现：

名称	地址	介绍
MCP协议	https://modelcontextprotocol.io/introduction	官方mcp协议以及主流语言实现
Awesome MCP Servers	https://mcpservers.org/	列出了多个可用的MCP服务器，涵盖了从生产级到实验性的各种实现，特别关注文件访问和工具集成。
Dify Connect MCP	https://marketplace.dify.ai/plugins/hjlarry/mcp-server	项目不仅提供完整的开源代码，还支持与MCP Inspector等工具的兼容性，极大地方便了开发者的调试和优化工作。可以让dify

…MCP 刚刚起步，很多生态还在逐渐完善。

04 MCP+LLM基本架构

Host（宿主）

宿主是嵌入 LLM 的应用程序，例如聊天机器人、虚拟助手、IDE 插件等。宿主负责确定何时需要从外部系统获取信息或执行操作，并将这些请求委托给 MCP 客户端。简单来说，宿主是 LLM 应用程序的载体，它利用 LLM 的能力，并通过 MCP 与外部世界进行交互。

Client（客户端）

客户端是位于宿主内部的组件，负责维护与 MCP 服务器的连接，并处理 MCP 协议的通信细节。客户端将宿主的请求转换为符合 MCP 协议的消息，并发送给相应的 MCP 服务器。同时，它还接收来自 MCP 服务器的响应，并将这些响应转换成宿主可以理解的格式。客户端可以看作是宿主与 MCP 服务器之间的桥梁。

Server（服务器）

服务器是独立的程序，负责提供对特定数据源或工具的访问。每个服务器都实现了 MCP 协议，并对外暴露一组定义好的功能。例如，一个数据库服务器可以提供查询数据库的功能，一个文件系统服务器可以提供读写文件的功能，一个 Web API 服务器可以提供访问特定 Web API 的功能。服务器抽象化了底层数据源和工具的复杂性，为客户端提供了一个统一的访问接口。

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

Resources：资源，类似于文件数据读取，可以是文件资源或是API响应返回的内容。资源是由服务器提供的、可以被客户端访问的数据单元。资源可以是静态的，例如文件、文档、图片等，也可以是动态的，例如数据库查询结果、API 响应等。与工具不同，资源通常是被动地提供数据，而不会执行任何操作。

Tools：工具，第三方服务、功能函数，通过此可控制LLM可调用哪些函数。工具是服务器提供的可执行的功能单元。每个工具都定义了其输入参数和输出结果，以及执行的具体逻辑。宿主可以通过客户端调用服务器提供的工具来执行特定的操作。例如，一个数据库服务器可以提供一个名为 “query” 的工具，用于执行 SQL 查询；一个文件系统服务器可以提供一个名为 “read” 的工具，用于读取文件内容。

Prompts：提示词，为用户预先定义好的完成特定任务的模板。mcp最主要的组件是客户端client和服务端server。但是在实际应用中，我们通常会在 LLM 的应用程序（host）利用mcp client，如对话工具、ide工具等，通过 MCP 协议（mcp-servers）进行通信。具体mcp sever如负责管理和连接业务存储、数据库、文档、web、外部数据等，可以重复利用，像积木一样开放给开发者共享。

那么MCP和Host如何组合，协调工作呢？每个host可能不太一样, 我只是梳理一个大概流程。

LLM要在真正使用MCP之前，要先明白LLML如何发现可用的 MCP，其次才是何时去调用MCP?

握手与能力协商：

首次握手：MCP Client 与每个 MCP Server 在连接初始化时互相鉴权（如 OAuth 或预共享密钥），并交换支持的功能列表和协议版本，确保双方兼容性。

**能力列表拉取******：握手完成后，Client 会对每个 Server 发起** tools/list、resources/list、prompts/list 等请求，获取该 Server 暴露的所有能力定义（名称、描述、JSON Schema）。

动态更新：若 Server 能力发生变化（如新增或下线工具），它可通过 notifications/tools/list_changed 主动通知 Client，Client 再次拉取最新列表，保证 LLM 始终掌握全量能力。

LLM 何时调用哪个 MCP？触发条件：意图识别与工具匹配。

**任务分析******：在生成用户回答时，LLM 内部会评估当前对话状态与工具元数据，判断是否存在“超出自身语言理解能力、需要外部操作”的场景，例如文件读写、API 查询或数据库操作等**。

**函数调用输出****：*一旦判定需要使用某工具，LLM 会按照类似 OpenAI Function Calling 的规范，在输出中生成一段 function_call JSON，指定 name（工具名）和* arguments（符合 inputSchema 的参数）。

调用流程：Client 转发与 Server 执行

请求封装：MCP Client 将 LLM 输出的调用指令封装为 JSON-RPC 格式的 CallToolRequest（或对应的 GetResourceRequest），并路由至相应的 MCP Server。

安全执行：MCP Server 在沙箱/容器中进行权限校验（RBAC、输入验证），执行真实操作（如调用 GitHub API、查询数据库、运行脚本），全程记录审计日志以满足合规需求。

**响应*******回传*******：操作完成后，Server 将结果封装为标准 JSON-RPC 响应，Client 再将其中有用部分（如文本、结构化数据等）注入到 LLM 的上下文，以便下一次推理使用。

05 MCP实战

现在开发mcp有很多方式，很多优秀框架比如FastMCP、LiteMCP 等（更多优秀框架），可以快速开发出mcp。本文以python为例，以Anthropic官方mcp为基础演示mcp client和mcp server开发模式。以简单实现一个实时获取天气为例。

1、环境准备

# 创建项目目录uv init clientcd client

 # 添加依赖uv add mcp openai python-dotenv httpx

2、编写MCP服务端

import jsonimport httpxfrom typing import Anyfrom mcp.server.fastmcp import FastMCPimport html_to_json# 初始化 MCP 服务器mcp = FastMCP("WeatherServer")OPENWEATHER_API_BASE = "https://www.tianqi24.com/{}/history{}{:0>2d}.html"USER_AGENT = "weather-app/1.0"async def fetch_weather(city: str, year: int, month: int) -> dict[str, Any] | None:    url = OPENWEATHER_API_BASE.format(city, year, month)    headers = {"User-Agent": USER_AGENT}    async with httpx.AsyncClient() as client:        try:            response = await client.get(url, headers=headers, timeout=30.0)            response.raise_for_status()            content = response.content.decode('utf-8')            return content  # 返回字典类型        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, city: str, year: int, month: int) -> str:    # 如果传入的是字符串，则先转换为字典    if isinstance(data, str):        try:            json_dict = html_to_json.convert(data)            x = \                json_dict['html'][0]['body'][0]['section'][0]['section'][0]['section'][0]['article'][1]['section'][0][                    'ul'][                    0][                    'li'][1:]            month_dict = {}            for ele in x:                _ = ele['div']                _month, _day = _[0]['_value'].split('-')                # assert _month == unify_dateStr(month) and len(_day) == 2                month_dict[_day] = {                    '天气': _[1]['_value'] if len(_[1]) == 1 else '转'.join(                        [_[1]['b'][0]['_value'], _[1]['span'][0]['_value'][2:]]),                    '最低温': _[3]['_value'].split('℃')[0],                    '最高温': _[2]['_value'].split('℃')[0],                    'AQI': _[4]['_value'],                    '风向': _[5]['_value'],                    '降雨量': _[6]['_value']                }        except Exception as e:            return f"无法解析天气数据: {e}"    # 如果数据中包含错误信息，直接返回错误提示    if "error" in data:        return f"⚠️ {data['error']}"    return (        f"🌍 {city}\n"        f"🌡 温度(最高): {month_dict[_day]['最高温']}°C\n"        f"🌡 温度(最): {month_dict[_day]['最高温']}°C\n"        f"💧 湿度: {month_dict[_day]['AQI']}%\n"        f"🌬 风速: {month_dict[_day]['风向']} m/s\n"        f"🌤 天气: {month_dict[_day]['降雨量']}\n"    )@mcp.tool()async def query_weather(city: str, year: int = 2025, month: int = 5) -> str:    """    输入指定城市的英文名称，返回今日天气查询结果。    """    data = await fetch_weather(city, year, month)    return format_weather(data)if __name__ == "__main__":    # 以标准 I/O 方式运行 MCP 服务器    mcp.run(transport='stdio')

@mcp.tool() 装饰器注册为 MCP 服务器的工具，使其能够被客户端调用。

MCP定义了Client与Server进行通讯的协议与消息格式，其支持两种类型通讯机制：标准输入输出通讯、基于SSE的HTTP通讯，分别对应着本地与远程通讯。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接口用于推送与接收数据。

所以上面的服务器代码无法直接运行，需要注意以下2点：

query_weather函数的函数说明至关重要，相当于是此后客户端对函数进行识别的基本依据，因此需要谨慎编写；

当指定 transport=‘stdio’ 运行 MCP 服务器时，客户端必须在启动时同时启动当前这个脚本，否则无法顺利通信。这是因为 stdio 模式是一种本地进程间通信（IPC，Inter-Process Communication）方式，它需要服务器作为子进程运行，并通过标准输入输出（stdin/stdout）进行数据交换。

3、客户端代码实现

import asyncioimport osfrom openai import OpenAIfrom dotenv import load_dotenvfrom contextlib import AsyncExitStackfrom mcp import ClientSession, StdioServerParametersfrom typing import Optionalfrom mcp.client.stdio import stdio_clientimport jsonimport sysload_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)        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()        print(response)        tools = response.tools        print("\n已连接到服务器，支持以下工具:", [tool.name for tool in tools])    async def process_query(self, query: str) -> str:        """调用 OpenAI API 处理用户查询"""        messages = [{"role": "system", "content": "你是一个智能助手，帮助用户回答问题。"},                    {"role": "user", "content": query}]        """        使用大模型处理查询并调用可用的 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__":    asyncio.run(main())

让LLM发现工具 →决定是否调用工具，以及哪个工具 → 返回工具名称和参数 → 执行工具 → 把结果作为新的上下文发给模型，生成最后代码返回给用户。

4、运行MCP

uv run client.py server.py

在mcp实际开发中，对开发者来说，不好调试，Anthropic提供了一个非常便捷的debug工具：Inspector。借助Inspector，能够非常快捷的调用各类server，并测试其功能。启动的时候，需要带上inspector。

npx-y @modelcontextprotocol/inspector uv run server.py

然后即可在本地浏览器查看当前工具运行情况：http://127.0.0.1:5173/[#resources](javascript:😉。

06 总结

除了stdio连接模式外，MCP还提供了可以服务器、客户端异地运行的SSE传输模式，以适用于更加通用的开发情况，以及现在逐渐推出可流式传输的 HTTP”来替代现有的 HTTP+SSE 方案。此举旨在解决当前远程 MCP 传输方式的关键限制，同时保留其优势。

MCP作为万能钥匙, 可以加速智能体的研发， MCP标准通信协议带来的最大价值之一，就是让广大Agent开发者能够基于此进行协作。已经诞生了数以千计的MCP服务器，允许用户直接下载并进行调用（见文末参考中的链接）。

MCP量身定制的Agent开发框架，通过集成MCP来提高Agent开发进度。未来在这些开源的努力，随着大模型应用更加成熟，相信在不久的将来，大模型应用场景会遍地开花。

如何学习大模型 AI ？

由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。

但是具体到个人，只能说是：

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

在这里插入图片描述