MCP（Model Context Protocol，模型上下文协议）由 Anthropic 于去年 11 月正式提出。

从本质上来说，MCP 是一种技术协议，是智能体（Agent）开发过程中共同约定的一种规范。这就好比秦始皇的“书同文、车同轨”，在统一的规范下，各方的协作效率得以大幅提高，最终提升智能体 Agent 的整体开发效率。

总的来说，MCP 解决的最大痛点，是 Agent 开发中调用外部工具的技术门槛过高的问题。

能够调用外部工具，是大模型进化为智能体 Agent 的关键。如果不能使用外部工具，大模型只能是一个简单的聊天机器人。由于底层技术限制，大模型本身无法直接与外部工具通信。因此，Function Calling 的思路是创建一个外部函数（Function）作为中介：一边传递大模型的请求，另一边调用外部工具，最终让大模型能够间接调用外部工具。

Function Calling 是一项非常出色的技术设计，自诞生以来一直被业内奉为圭臻。但其存在的问题是，编写外部函数的工作量过大。一个简单的外部函数往往需要上百行代码；为了让大模型“认识”这些函数，还需要额外为每个函数编写 JSON Schema 格式的功能说明；此外，还需精心设计提示词模板，才能提高 Function Calling 响应的准确率。

MCP 的目标，就是在 Agent 开发过程中，让大模型更加便捷地调用外部工具。

为了让大模型更便捷地调用外部工具，MCP 提出了“车同轨、书同文”的方案，统一 Function Calling 的运行规范：

MCP 将大模型运行环境称作 MCP Client（MCP 客户端），将外部函数运行环境称作 MCP Server（MCP 服务器）。

统一 MCP 客户端和服务器的运行规范，并要求双方按照既定的提示词模板进行通信。

这种规范最大的好处在于，可以避免 MCP 服务器的重复开发。例如，查询天气、网页爬取、查询本地 MySQL 数据库等通用需求，只需一人开发好服务器，其他人即可直接复制到项目中使用，无需每次都单独编写。

“车同轨、书同文”的规范极大地促进了全球 AI 开发者的协作。很快，GitHub 上出现了海量的 MCP 服务器，涵盖 SQL 数据库检索、网页浏览与爬取、命令行操作、数据分析与机器学习建模等各个领域。

现在，只要本地运行的大模型支持 MCP 协议（即安装了相关库），仅需几行代码即可接入这些海量外部工具，Agent 的开发门槛瞬间降低。

这种旨在提高大模型 Agent 开发效率的技术协议，与 HTTP 等网络信息交换协议的目标一致，都是通过提高协作效率来提升开发效率。

MCP 客户端不仅支持 Claude 模型，也支持任意本地模型、在线大模型或 IDE。例如，Cursor 正式接入 MCP 后，不仅能快速编写 MCP 服务器，更能一键连接成百上千的开源 MCP 服务器，让大模型快速接入海量工具，大幅加快 Agent 开发进度。

以下是一个基础的 MCP 客户端实现代码：

import asyncio
import os
import json
from typing import Optional
from contextlib import AsyncExitStack

from openai import OpenAI  
from dotenv import load_dotenv

from mcp import ClientSession, StdioServerParameters
from 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 服务器并建立通信
        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()
        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())

使用 AsyncExitStack 来管理异步上下文，确保 MCP 会话和传输层在程序退出或异常时能够被正确清理。

通过 dotenv 加载 .env 文件中的配置（如 OPENAI_API_KEY、BASE_URL、MODEL），避免硬编码敏感信息。

在 process_query 中，每次处理请求前都会调用 session.list_tools() 动态获取服务器当前可用的工具列表，并自动转换为 OpenAI Function Calling 所需的 tools 格式。

当大模型返回 finish_reason == "tool_calls" 时，客户端会自动解析参数，调用 session.call_tool 执行工具，并将结果作为 tool 角色消息追加到上下文中，再次请求大模型生成最终回复。

在 main 函数中校验命令行参数，确保传入了服务器脚本路径，并统一在文件顶部导入 sys 模块，避免运行时错误。

根据 MCP 协议定义，Server 可以提供三种类型的标准能力。每个 Server 可同时提供这三种能力，或仅提供其中一种：

：类似于文件数据读取，可以是本地文件资源，也可以是 API 响应返回的内容。

：第三方服务或功能函数，通过此能力控制 LLM 可调用哪些函数。

：为用户预先定义好的、用于完成特定任务的模板。

Model Context Protocol（MCP）旨在将大型语言模型直接连接至数据源，实现无缝集成。根据 MCP 规范，当前支持两种传输方式：

SSE 仅支持服务器向客户端的单向通信，而 WebSocket 则支持双向通信。HTTP 采用传统的请求-响应模式，每次请求通常建立新的连接；SSE 则基于长连接，适用于实时通知、股票行情更新等场景。

创建一个标准的 MCP 服务器通常包含以下步骤：

引入 mcp.server.fastmcp 模块并初始化 FastMCP 实例。

定义具体的业务逻辑函数（如 API 请求、数据处理等）。

使用 @mcp.tool() 装饰器将业务函数注册为 MCP 工具，并编写清晰的文档字符串（Docstring）作为工具描述。

在 __main__ 中调用 mcp.run(transport='stdio') 启动服务器。

以下是一个完整的MCP 服务器实现：

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 = "YOUR_API_KEY"  # 请替换为你自己的 OpenWeather API Key
USER_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__":
    # 以标准 I/O 方式运行 MCP 服务器
    mcp.run(transport='stdio')

使用 FastMCP 可以极大地简化服务器的创建过程，无需手动处理底层的 JSON-RPC 通信协议。

@mcp.tool() 装饰器会自动将函数名和参数类型转换为 MCP 工具定义。函数的 Docstring（如“输入指定城市的英文名称…”）至关重要，它会被大模型读取并作为工具的功能说明，直接影响大模型调用工具的准确率。

使用 httpx.AsyncClient 进行异步 HTTP 请求，避免阻塞 MCP 服务器的事件循环，确保高并发下的响应性能。

在 fetch_weather 中捕获了 HTTPStatusError 和通用 Exception，并将错误信息封装为字典返回，防止因外部 API 异常导致整个 MCP 服务器崩溃。

将原始的 JSON 数据通过 format_weather 转换为人类可读的文本格式，这有助于大模型更好地理解工具返回的结果，从而生成更准确的回复。

深入理解 MCP 协议：从原理到工程实践

人工智能

编程语言

开发工具

MCP协议由Anthropic提出，通过统一Function Calling规范，解决Agent开发中工具调用的高门槛问题。文章详解MCP客户端与服务器开发流程，包含代码实现、stdio/SSE传输机制，以及如何快速接入外部工具提升智能体开发效率。

聊天机器人

SQL数据库

机器学习

数据分析

数据处理

数据库

服务器

WebSocket

智能体

GitHub

MySQL

Agent

JSON

腾讯云数据库特惠，新用户首单0.5折起


文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

EdgeOne AI 安全实战专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

深入理解 MCP 协议：从原理到工程实践

深入理解 MCP 协议：从原理到工程实践

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐