MCP协议

MCP是什么

MCP即模型上下文协议（Model Context Protocol），是一种开放协议，旨在为大型语言模型（LLM）提供标准化的上下文访问方式。官方举例说，MCP犹如AI应用的 “USB-C端口”，使LLM能够通过统一的接口无缝连接至多种数据源和工具（如文件系统、数据库或外部API）。它是基于JSON-RPC格式构建的，提供了一种面向客户端和服务器之间上下文交换和采样协调的有状态会话协议。正逐渐成为事实标准。

与FunctionCall的区别与联系

FunctionCall工作流程

FunctionCall指的是AI模型根据上下文自动执行函数的机制，用来规定模型是如何调用函数的。
FunctionCall一般的过程如下：

1. 将用户的自然语言输入与已有函数的描述作为输入参数传给LLM。
2. LLM结合输入参数，决定调用哪些函数，并指明必要参数（如函数的入参），进行格式化（如JSON、XML格式）的输出。
3. 用户端接收到LLM格式化的函数调用后，对本地的函数进行调用，得到结果。
4. 将得到的函数结果传给LLM，使得LLM有了所需的上下文信息。

FunctionCall实际上强调的是LLM本身的能力，一些经过特殊训练或调优的LLM能够根据用户的自然语言输入决定使用哪些函数，并按约定的格式表达出函数的调用。这里所描述的 “格式”，不同LLM提供商之间是可能有差异的。

// OpenAI GPT的function call 格式
{
    "type": "function_call",
    "id": "fc_12345xyz",
    "call_id": "call_12345xyz",
    "name": "get_weather",
    "arguments": "{\"location\": \"Shanghai\"}"
}

// Google Gemini的function call 格式
{
  "functionCall": {
    "name": "get_weather",
    "args": {
      "location": "Shanghai"
    }
  }
}

FunctionCall的缺点

FunctionCall的缺点在于处理不好多轮对话和复杂需求，适合边界清晰、描述明确的任务。如果需要处理很多的任务，那么FunctionCall的代码比较难维护。
需要自行维护可用函数列表、外部系统的接入需要进行针对适配，不具有通用性。

MCP与FunctionCall的联系与区别

两者具有一定联系，但侧重点不同。
FunctionCall侧重于描述LLM本身具有的结构化函数调用能力（即调用哪些函数），而MCP则侧重于描述函数的规范化执行（即怎么执行被调用的函数），大概是分配与执行的关系。具体区别如下表所示：

比较维度	FunctionCall	MCP
性质	功能	协议
主要职责	解析用户意图并选择合适的函数调用，并进行格式化输出	规范化函数的具体执行过程，即规范LLM应用与外部系统的交互
责任方	各LLM提供商，平台依赖性强	LLM应用（client端）、外部系统（server端），LLM兼容
数据结构	因LLM提供商而有所不同，缺乏统一标准	规范的JSON-RPC
灵活性	调用预定义函数，功能扩展需要额外开发	允许运行时发现新工具，动态适配和扩展
使用场景	轻量级、单任务、封闭环境应用	复杂场景、多数据源、跨平台协作

两者不是替代关系。在实际应用中，MCP和FunctionCall相互配合，发挥各自的优势。
例如，某教育智能体采用 “MCP + FunctionCall” 组合：使用MCP调用题库API生成题目，通过FunctionCall调用本地Markdown渲染工具格式化答案。

目前MCP的缺点

1. 生态尚不足，支持MCP的应用数量较少
2. 很难判断MCP服务器的安全性
3. 部署和管理有一定门槛、产品化程度不高

MCP架构、工作流程

MCP架构

MCP采用客户端 - 服务器模型，其核心组件包括MCP Hosts、MCP Clients和MCP Servers：

1. MCP Hosts：运行MCP的主应用程序，例如Claude Desktop、Cursor或AI Agent，负责通过MCP访问外部数据。
2. MCP Clients：协议客户端，包含于Host中，由其管理。客户端与服务器建立一对一连接，负责发送请求并接收响应。
3. MCP Servers：轻量级程序，通过MCP暴露特定功能（如文件读取或API调用）。

MCP工作流程

MCP的基本工作流程如下：

1. 客户端启动，读取MCP配置：客户端在启动时会读取相关的MCP配置信息，这些信息可能包括MCP Server的地址、可用工具的相关配置等，为后续的连接和操作做准备。
2. 连接MCP Server：客户端根据读取的配置信息，尝试与MCP Server建立连接。在连接过程中，客户端可能会向Server发送连接请求，并等待Server的响应，以确认连接是否成功。
3. 获取工具列表：连接成功后，客户端向MCP Server请求获取可用的工具列表。Server会将支持的工具信息返回给客户端，客户端可以根据这些信息了解有哪些工具可供使用。
4. 发送用户问题时，同时提供可用工具列表：当用户提出问题时，客户端会将用户的问题以及可用的工具列表一起发送给LLM。这样LLM在处理用户问题时，可以根据工具列表来判断是否需要调用工具。
5. LLM判断是否需要调用工具：LLM接收到用户问题和工具列表后，会根据问题的内容和自身的逻辑判断是否需要调用工具来解决问题。如果需要调用工具，LLM会指定需要调用的工具和相应的参数。
6. 客户端按照LLM指示调用Server上的工具：客户端接收到LLM的指示后，会根据指示调用MCP Server上相应的工具。客户端将调用请求发送给Server，Server执行工具并将结果返回给客户端，客户端再将结果反馈给LLM，最终由LLM生成用户问题的最终回复。

结合mcpserver.py和mcpclient.py代码片段可以进一步了解MCP的工作流程。

import json
import httpx
from typing import Any
from mcp.server.fastmcp import FastMCP
 
# 初始化 MCP 服务器
mcp = FastMCP("WeatherServer")
 
# 天气 API 配置
OPENWEATHER_API_BASE = "https://api.seniverse.com/v3/weather/now.json"
API_KEY = "****"  # 请替换为你自己的天气 API Key
USER_AGENT = "weather-app/1.0"
 
# 从天气 API 获取天气信息
async def fetch_weather(city: str) -> dict[str, Any] | None:
    """
    从 天气 API 获取天气信息。
    :param city: 城市名称（需使用英文，如 beijing）
    :return: 天气数据字典；若出错返回包含 error 信息的字典
    """
    params = {
        "location": city,
        "key": API_KEY,
        "language": "zh-Hans",
        "unit": "c"
    }
 
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(OPENWEATHER_API_BASE, params=params, timeout=30.0)
            print(response)
            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 "status_code" in data:
        return f"⚠️ {data['error']}"
    # 提取数据时做容错处理
    result = data.get("results", [{}])[0]  # 若results为空则返回空字典
 
 
    # 解析location信息（带多层防护）
    location = result.get("location", {})
    city = location.get("name", "未知")
    country = location.get("country", "未知")
 
    # 解析now对象（类型转换增强）
    now = result.get("now", {})
    temp = now.get("temperature", "N/A")  # 保持字符串类型，如需数值可做转换
    description = now.get("text", "未知")
 
    # 处理旧版接口字段（新版无对应字段）
    humidity = "N/A"  # 新版数据结构无湿度字段
    wind_speed = "N/A"  # 新版数据结构无风速字段
 
    return (
        f"🌍 {city}, {country}\n"
        f"🌡 温度: {temp}°C\n"
        f"💧 湿度: {humidity}%\n"
        f"🌬 风速: {wind_speed} m/s\n"
        f"🌤 天气: {description}\n"
    )
 
# 装饰器，用于将函数注册为 MCP 服务器对外提供的工具
@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')

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和input_schema是MCP服务器上的工具的描述和输入参数
                "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(result)
            print(f"\n\n[Calling tool {tool_name} with args {tool_args}]\n\n")
            print("#########################")
            print(result.content[0].text)
 
            # 将模型返回的调用哪个工具数据和工具执行完成后的数据都存入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("\nuser: ").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())

写在最后

MCP协议主要规定了两个部分的内容：

1. 动态地提供对可用函数的标准化的描述，每个Server有哪些函数可以用；
2. 标准化对外部系统的调用与结果的处理，如何调用这些函数。

这两点可以总结为函数的注册和使用，让大模型感知外部环境的一个协议。
MCP本身并没有规定与大模型的交互方式，不同的MCP Host与大模型交互会有差异，例如提示词方式与FunctionCall方式。

---

开发生态