本文介绍了Anthropic公司提出的MCP（Model Context Protocol）模型上下文协议，旨在为大模型Function Calling提供统一的标准。MCP类似于计算机世界的USB接口，规范了应用程序向LLM提供上下文的方式，使AI模型能够连接到不同的外部数据源和工具。文章通过一个简单的加法计算案例，演示了如何使用Python实现MCP的客户端和服务器端，展示了MCP协议在实际应用中的价值，即实现大模型与外部世界的交互。

💡MCP是一个标准化开放协议，旨在规范应用程序向LLM提供上下文的方式，类似于计算机世界的USB接口，让大模型Function Calling有统一的标准可循。

🧩MCP架构包含三个核心角色：客户端、服务端和资源。外部应用和数据源通过MCP协议开放接口能力，AI应用通过MCP协议接入外部工具和数据源，实现LLM大模型与外部世界的交互。

🔑客户端代码的关键在于`connect_to_server`方法，它负责连接到MCP服务器并列出可用的工具函数。在与大模型交互时，客户端会将查询发送给大模型，并根据大模型的建议调用相应的工具。

🔄在客户端与大模型的交互中，如果大模型返回的结果是使用工具的建议，客户端会解析并调用工具，然后将工具的执行结果返回给大模型，用于生成最终结果。这个过程与Function Calling的原理类似，但MCP提供了一套标准化的流程。

1.引言

上一篇文章，我们知道function calling实现了大模型通往外部世界的桥梁，于大模型而言，function calling实现了一种插件化的机制，完成能力扩展。

你还记得编程实现function calling有哪些关键点吗？首先要定义一个工具函数，像这样：

其次，要通过dict数据结构形式描述工具，这很关键（大模型根据描述来选择工具），像这样：

我们看到，要实现function calling是一件相对繁琐的事情，而且这里还存在巴比塔困境，即各家大模型各玩各的，简单直白说，就是没有规矩，不成方圆。

于是，2024年11月Anthropic（发布Claude大模型）公司，提出了mcp（Model Context Protocol）模型上下文协议，试图通过标准化协议方式，实现书同文，车同轨。

这是一种进步，任何事物的发展，当从百花齐放到一统江湖的时候，说明黎明的曙光到来。接下来我将通过两篇文章，手把手给你分享mcp两种通信模式的实现。

2.案例

2.1.什么是mcp

mcp是一个标准化开放协议，规范了应用程序向LLM提供上下文的方式。类似于计算机世界USB接口，mcp提供了一种标准化的方式让AI模型能够连接到不同的外部数据源和工具。简单说，就是让大模型function calling有统一的标准可循。

mcp架构图：c/s 机构模式

从架构图中，mcp有三个核心角色：

客户端服务端资源

实践应用流程图：

外部应用和数据源，通过mcp协议开放接口能力，AI应用通过mcp协议接入外部工具和数据源，实现LLM大模型于外部世界的交互。

访问：mcp.so/，看到很多大的玩家已经通过mcp开放了相关接口服务能力

2.2.环境准备

搞清楚了mcp相关的概念和流程，我们自己从代码层面实现一套mcp完整流程。当然，要先安装相关的库准备环境

pip install uv mcp

备注一下，只安装mcp库其实就够用了，网上很多文章都绑定了uv库，uv等价于python中pip+虚拟环境，不用也是可以的。

2.3.server端代码

server端实现工具接口，数据开放。等价于function calling中函数定义

from mcp.server.fastmcp import FastMCP# 创建MCP 服务mcp = FastMCP('Demo')@mcp.tool()def add(a:int, b:int) ->int:  """"  计算两个整数的和并返回  """  return a+bif __name__ == "__main__":  # 以标准 I/O 方式运行 MCP 服务器  mcp.run(transport='stdio')

2.4.客户端代码

import asyncioimport jsonfrom typing import Optionalfrom contextlib import AsyncExitStackfrom openai import OpenAIfrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientclass MCPClient:    def __init__(self):        """初始化MCP客户端"""        self.exit_stack = AsyncExitStack()        self.opanai_api_key = ""  # 调用模型的api_key        self.base_url = "https://api.deepseek.com"  # 调用模型url, 这里以deepseek作演示        self.model = "deepseek-chat"  # 调用deepseek-v3模型        self.client = OpenAI(api_key=self.opanai_api_key, base_url=self.base_url)        self.session: Optional[ClientSession] = None  # Optional提醒用户该属性是可选的，可能为None        self.exit_stack = AsyncExitStack()  # 用来存储和清除对话中上下文的，提高异步资源利用率    async def connect_to_server(self, server_script_path):        """连接到MCP服务器并列出MCP服务器的可用工具函数"""        server_params = StdioServerParameters(            command="python",            args=[server_script_path],            env=None        )  # 设置启动服务器的参数, 这里是要用python执行server.py文件        # 启动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()  # 与服务器建立stdio连接        # 列出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) -> 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,                "input_schema": 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}]\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)                print(f"\nDeepSeek-V3: {response}")            except Exception as e:                print(f"发生错误: {str(e)}")    async def clean(self):        """清理资源"""        await self.exit_stack.aclose()async def main():    client = MCPClient()    try:        await client.connect_to_server(sys.argv[1])        await client.chat_loop()    finally:        await client.clean()if __name__ == "__main__":    import sys    asyncio.run(main())

客户端实现连接，协议转换，重点要关注connect_to_server方法

在与大模型交互中，有一段很熟悉的代码

这不就是我们function calling吗？本质没有变化，mcp只是给我们提供了一套标准，让大家在一个统一的标准下玩，便于开放互联互通。

2.5.执行结果

代码都开发完了，启动执行看看结果，需要注意，我们这篇文章演示标准IO模式，即mcp的stdio本地通信模式，服务端以子进程方式运行，命令如下：

python mymcp/hello_mcp_client.py mymcp/hello_mcp_server.py

启动完成，我们看到成功列出服务端开放的工具列表，这里是add工具

输入信息：111求和222等于多少，与大模型愉快交流

大模型理解我们的问题，选择了add工具，最终完成任务。