为何选择MCP?自建流程与Anthropic MCP的对比分析

发布于:2025-05-13 ⋅ 阅读:(17) ⋅ 点赞:(0)

引言

在当前的AI技术浪潮中,如何高效、可靠地让大型语言模型(LLM)与外部工具和服务进行交互,是一个核心议题。Anthropic提出的模型组件协议(Model Component Protocol, MCP)旨在为此提供一个标准化的解决方案。然而,一个关键问题随之而来:采用MCP与我们自行构建一套类似的工具调用流程,究竟有何本质区别?我们选择MCP的驱动力是什么? 本文将深入剖析这两者之间的异同,并探讨MCP背后的潜在意义。

初识MCP:标准的价值与额外的负担?

在这里插入图片描述

MCP(模型上下文协议) 是 Anthropic 在 2024 年 11 月底推出的一个开放标准,它的作用是统一大型语言模型(LLM)和外部数据源、工具之间的通信方式。MCP 主要是为了解决当前 AI 模型因为数据孤岛问题,没办法充分发挥能力的难题。有了 MCP,AI 应用就能安全地访问和操作本地以及远程的数据,还为 AI 应用提供了连接各种事物的接口。

初次接触MCP时,许多人的直观感受可能是:这似乎是大厂为了构建自身生态、争夺行业话语权而推出的标准。它在现有的模型与工具交互流程中引入了一系列新概念(如MCP服务端、客户端),无疑增加了开发者的理解成本和认知负担。那么,从纯粹的技术实现角度看,它与自建流程真的不同吗?

流程对比:自建方案 vs. MCP方案

为了更清晰地说明问题,我们来分解一下两种方案的具体流程:

1. 自建工具调用流程:

  • 提示词工程: 工程师为每个所需工具精心设计提示词(Prompt)。
  • 后端实现:
    • 开发模型调用逻辑,集成预设的工具提示词。
    • 根据模型返回的工具调用请求(通常遵循类似Function Calling的格式),解析参数。
    • 实现调用相应外部API或内部函数的逻辑。
    • 将API返回结果再次交给模型进行处理和总结。
  • 接口提供: 后端向前端或其他调用方提供统一的API接口。
  • 执行过程:
    • 前端传入用户查询(Query)。
    • 后端调用AI模型,模型根据Query和工具提示词,判断是否需要使用工具及所需参数。
    • 后端执行工具调用(调用API/函数),获取数据。
    • 将数据返回给AI模型,生成最终回复。
    • 流式(或一次性)返回给用户。

2. MCP工具调用流程:

  • 提示词工程: 同样需要工程师为每个工具设计提示词。
  • 后端实现:
    • 开发MCP服务端:负责实际执行工具(调用API/函数),管理工具的具体实现逻辑。
    • 开发MCP客户端
      • 负责与AI模型交互,根据用户Query和工具提示词获取意图,确定要使用的工具和参数。
      • 向前端提供API接口。
  • 执行过程:
    • 前端传入用户查询(Query)。
    • MCP客户端调用AI模型,获取需要使用的工具及其参数。
    • MCP客户端将确定的工具名称和参数发送给MCP服务端
    • MCP服务端根据收到的请求,执行相应的工具API调用,并将结果返回给MCP客户端
    • MCP客户端接收到服务端的执行结果后,再次调用AI模型,让模型基于此结果生成最终回复。
    • 流式(或一次性)返回给用户。

MCP服务端实践:以SSH管理工具为例

为了更具体地理解MCP的实现方式,让我们看一个实际的MCP服务端代码示例。这个示例使用Python的FastMCP库构建了一个名为SSHMCP的服务端,其核心功能是提供一系列管理SSH连接的工具:

# (此处仅展示关键部分)
import paramiko
from mcp.server.fastmcp import FastMCP

# 1. 初始化MCP服务
mcp = FastMCP('SSHMCP')

# 2. 定义工具 (以 execute_ssh_command 为例)
@mcp.tool(
    name='execute_ssh_command',  # 工具名称,供AI识别
    description='在已连接的SSH客户端上执行命令并返回结果' # 工具描述,供AI理解功能
)
def execute_ssh_command(client_key: str, command: str) -> str: # 参数定义,供AI填充
    """
    在已连接的SSH客户端上执行命令并返回结果
    (函数体包含具体实现逻辑,使用paramiko执行SSH命令)
    """
    global ssh_clients
    client = ssh_clients.get(client_key)
    if not client:
        print(f"错误: SSH客户端 {client_key} 未连接")
        return None
    try:
        print(f"执行命令: {command}")
        stdin, stdout, stderr = client.exec_command(command)
        result = stdout.read().decode()
        error = stderr.read().decode()
        if result: print(f"命令输出:\n{result}")
        if error: print(f"错误信息:\n{error}")
        return result # 通常只返回主要结果给AI
    except Exception as e:
        print(f"命令执行失败: {str(e)}")
        return f"命令执行失败: {str(e)}" # 返回错误信息

# 其他工具如 add_ssh_config, list_ssh_configs 等也使用 @mcp.tool 定义

# 3. 启动服务
# if __name__ == "__main__":
#     port = 9055
#     app = mcp.sse_app()
#     uvicorn.run(app, port=port, host='0.0.0.0')

从示例代码看MCP的特点:

  1. 标准化工具定义: 通过@mcp.tool装饰器,开发者可以清晰地将一个Python函数注册为MCP工具。关键在于提供明确的name(工具的唯一标识符)和description(自然语言描述,极其重要,因为这是AI模型理解工具功能的依据)。
  2. 参数化接口: 函数的参数(如execute_ssh_command中的client_keycommand)及其类型提示(str)定义了调用该工具需要提供的信息。MCP客户端中的AI模型需要根据用户意图,解析出这些参数的值。
  3. 封装具体实现: 每个被@mcp.tool装饰的函数内部,包含了执行该工具所需的具体业务逻辑(例如,使用paramiko库进行SSH连接、命令执行、配置管理等)。这部分逻辑对MCP客户端是透明的,客户端只需知道工具名称和所需参数即可调用。
  4. 服务端职责清晰: 这个例子完美诠释了MCP服务端的角色——作为一系列能力的提供者。它维护自身状态(如ssh_configs, ssh_clients),并暴露标准化的工具接口供客户端(间接通过AI)调用。

与流程对比的联系:

这个服务端示例,恰好对应了我们之前讨论的MCP流程中的“MCP服务端”部分。当MCP客户端从AI获取到需要调用如execute_ssh_command工具及参数{'client_key': 'my_server', 'command': 'ls -l'}的指令后,它会将这个请求(工具名+参数)发送给这个正在运行的服务端。服务端接收到请求后,查找名为execute_ssh_command的工具函数,传入参数并执行,最终将执行结果(如ls -l的输出)返回给客户端。

MCP客户端实践:以Cherry Studio为例

在讨论了MCP服务端的实现之后,我们再来看看MCP客户端是如何与服务端进行交互和利用其提供的工具的。这里我们以Cherry Studio这款应用为例,结合其界面截图进行说明。

  1. 配置连接MCP服务器 (图1)

正如第一张图片所示,Cherry Studio作为MCP客户端应用,提供了管理MCP服务器连接的功能(界面左侧导航栏中的“MCP 服务器”)。

  • 添加与配置: 用户可以在这里添加新的MCP服务器配置。
  • 指定地址与类型: 关键配置在于指定MCP服务器的URL(例如截图中显示的 http://0.0.0.0:9055/sse,这恰好与我们之前Python SSH MCP服务端的监听地址和端口一致)和类型(如“服务器发送事件 (sse)”,表明客户端将使用Server-Sent Events协议与该服务器通信)。
  • 启用连接: 通过总开关,可以方便地启用或禁用整个MCP服务器的连接。

这一步是建立客户端与服务端联系的基础。客户端需要知道服务端的“地址”和“沟通方式”。

  1. 发现与管理可用工具

连接建立后,客户端会与服务器通信以发现其提供的工具。第二张图展示了在设置界面的“工具”标签下,Cherry Studio列出了从服务端(我们之前的SSH示例)发现的所有可用工具,包含其名称和描述(如 execute_ssh_command - “在已连接的SSH客户端上执行命令并返回结果”)

  1. mcp在对话中的集成与可用性

下图展示的是在应用的主交互界面(对话框)中,配置好的、且在设置中被启用的MCP工具。

  • AI感知: 参与对话的AI模型现在已经感知到了这些来自外部MCP服务器的工具(如SSH管理工具)。
  • 调用基础: 当用户在对话框中提出需要操作远程服务器的请求时(例如,“帮我列出服务器A上的文件”或“在服务器B上运行某个脚本”),AI模型就可以根据这些工具的描述信息,判断出可以使用如 list_ssh_configs 来查找服务器A的配置名,或使用 execute_ssh_command 工具来执行具体命令。
  • 无缝集成: MCP使得这些外部工具能够相对无缝地集成到AI的核心交互流程中,扩展了AI在对话场景下能完成的任务范围。

核心流程的相似性

通过对比流程和审视MCP服务端的具体实现,我们可以再次确认:无论是自建还是采用MCP,其核心逻辑链条

AI理解意图 -> AI决定调用工具 -> (通过某种机制)执行工具 -> AI基于工具结果生成回复

是高度一致的。MCP引入了客户端/服务端的明确划分和标准化的工具定义接口(如@mcp.tool),但这更像是一种架构上的规范和选择,而非流程本身的根本性变革。

MCP是必需品吗?

有人可能会提出,MCP的价值在于其标准化协议,能够方便不同企业间的MCP客户端和服务端进行互操作。例如,我们的MCP客户端可以方便地调用其他企业的MCP服务端。

然而,这种观点值得商榷。首先,即使不使用MCP,只要遵循目前主流的Function Calling或类似的模型工具调用规范(其参数格式已趋于稳定和明确),不同系统间的对接差异本身就相对较小。其次,MCP协议本身并未完全消除这种差异性;实际集成时,客户端往往仍需根据特定服务端的细节进行适配。因此,仅仅为了互操作性而选择MCP,理由似乎并不充分。

MCP的真正价值与深层考量

如果MCP在技术流程上没有带来颠覆性改变,其互操作性优势也并非不可替代,那么Anthropic力推MCP的真正意图是什么?

  1. 标准化愿景与潜力: 在当前AI技术“诸神混战”、上下游充满不确定性的时代,MCP提出了一套技术标准化的协议。这符合行业对规范化、标准化的期待,是共建未来AI生态愿景的一部分。从长远看,一个被广泛接受的标准无疑具有巨大潜力。
  2. 生态控制权与行业话语权: 这或许是更深层次的原因。MCP表面上是一个开放协议,旨在解决AI模型与外部工具集成的碎片化问题,但其背后,也体现了Anthropic(以及其他参与者)对未来AI生态主导权的争夺。如果MCP最终成为行业共识的标准协议,那么围绕该协议,甚至可能统一各大模型厂商的工具调用格式。届时,作为标准制定和推广的核心参与者,Anthropic在行业中的地位和影响力将不言而喻。

完整代码

import paramiko
import uvicorn
import json
import os
from typing import Dict, List, Optional
from mcp.server.fastmcp import FastMCP

mcp = FastMCP('SSHMCP')

# 添加全局变量存储SSH连接和配置
ssh_clients = {}
ssh_configs = {}
CONFIG_FILE = "ssh_configs.json"

# 加载已保存的SSH配置
def load_ssh_configs():
    global ssh_configs
    if os.path.exists(CONFIG_FILE):
        try:
            with open(CONFIG_FILE, 'r') as f:
                ssh_configs = json.load(f)
            print(f"已加载 {len(ssh_configs)} 个SSH配置")
        except Exception as e:
            print(f"加载SSH配置失败: {str(e)}")
            ssh_configs = {}
    else:
        ssh_configs = {}

# 保存SSH配置到文件
def save_ssh_configs():
    try:
        with open(CONFIG_FILE, 'w') as f:
            json.dump(ssh_configs, f, indent=2)
        print("SSH配置已保存")
    except Exception as e:
        print(f"保存SSH配置失败: {str(e)}")

# 初始化时加载配置
load_ssh_configs()


@mcp.tool(
    name='delete_ssh_config',
    description='删除SSH连接配置'
)
def delete_ssh_config(name: str) -> bool:
    """
    删除SSH连接配置
    
    Args:
        name: 配置名称
    
    Returns:
        是否成功删除配置
    """
    global ssh_configs
    
    if name not in ssh_configs:
        print(f"配置名称 '{name}' 不存在")
        return False
    
    # 如果该配置有活跃连接,先断开
    client_key = f"{name}"
    if client_key in ssh_clients:
        ssh_disconnect(client_key)
    
    # 删除配置
    del ssh_configs[name]
    save_ssh_configs()
    print(f"已删除SSH配置: {name}")
    return True

@mcp.tool(
    name='test_ssh_connection',
    description='测试SSH连接配置是否可用'
)
def test_ssh_connection(hostname: str, port: int = 22, username: str = None, 
                      password: str = None, key_filename: str = None) -> bool:
    """
    测试SSH连接配置是否可用
    
    Args:
        hostname: SSH服务器地址
        port: SSH端口,默认为22
        username: 用户名
        password: 密码(如果使用密码认证)
        key_filename: SSH私钥文件路径(如果使用密钥认证)
    
    Returns:
        连接是否成功
    """
    # 创建临时SSH客户端
    client = paramiko.SSHClient()
    
    # 自动添加策略,保存服务器的主机名和密钥信息
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    
    try:
        # 根据提供的认证方式连接SSH服务器
        if key_filename:
            # 使用SSH密钥认证
            client.connect(hostname=hostname, port=port, username=username, key_filename=key_filename, timeout=10)
            print(f"测试连接成功: 已成功使用SSH密钥连接到 {hostname}:{port}")
        else:
            # 使用密码认证
            client.connect(hostname=hostname, port=port, username=username, password=password, timeout=10)
            print(f"测试连接成功: 已成功使用密码连接到 {hostname}:{port}")
        
        # 关闭连接
        client.close()
        return True
    
    except Exception as e:
        print(f"测试连接失败: {str(e)}")
        return False

@mcp.tool(
    name='add_ssh_config',
    description='添加SSH连接配置(会先测试连接)'
)
def add_ssh_config(name: str, hostname: str, port: int = 22, username: str = None, 
                  password: str = None, key_filename: str = None) -> bool:
    """
    添加SSH连接配置(会先测试连接)
    
    Args:
        name: 配置名称
        hostname: SSH服务器地址
        port: SSH端口,默认为22
        username: 用户名
        password: 密码(如果使用密码认证)
        key_filename: SSH私钥文件路径(如果使用密钥认证)
    
    Returns:
        是否成功添加配置
    """
    global ssh_configs
    
    if name in ssh_configs:
        print(f"配置名称 '{name}' 已存在,请使用其他名称或先删除现有配置")
        return False
    
    # 先测试连接
    print(f"正在测试连接 {hostname}:{port}...")
    if not test_ssh_connection(hostname, port, username, password, key_filename):
        print(f"添加配置失败: 连接测试未通过")
        return False
    
    # 创建新的配置
    config = {
        "hostname": hostname,
        "port": port,
        "username": username,
        "password": password,
        "key_filename": key_filename
    }
    
    # 添加到配置字典
    ssh_configs[name] = config
    save_ssh_configs()
    print(f"已添加SSH配置: {name}")
    return True

@mcp.tool(
    name='update_ssh_config',
    description='修改SSH连接配置(会先测试连接)'
)
def update_ssh_config(name: str, hostname: str = None, port: int = None, 
                     username: str = None, password: str = None, key_filename: str = None) -> bool:
    """
    修改SSH连接配置(会先测试连接)
    
    Args:
        name: 配置名称
        hostname: SSH服务器地址
        port: SSH端口
        username: 用户名
        password: 密码(如果使用密码认证)
        key_filename: SSH私钥文件路径(如果使用密钥认证)
    
    Returns:
        是否成功修改配置
    """
    global ssh_configs
    
    if name not in ssh_configs:
        print(f"配置名称 '{name}' 不存在")
        return False
    
    # 获取现有配置
    old_config = ssh_configs[name].copy()
    
    # 准备新配置(先复制旧配置,然后更新非空参数)
    new_config = old_config.copy()
    if hostname is not None:
        new_config["hostname"] = hostname
    if port is not None:
        new_config["port"] = port
    if username is not None:
        new_config["username"] = username
    if password is not None:
        new_config["password"] = password
    if key_filename is not None:
        new_config["key_filename"] = key_filename
    
    # 测试新配置
    print(f"正在测试新配置的连接...")
    if not test_ssh_connection(
        new_config["hostname"], 
        new_config["port"], 
        new_config["username"], 
        new_config["password"], 
        new_config["key_filename"]
    ):
        print(f"修改配置失败: 新配置连接测试未通过")
        return False
    
    # 测试通过,更新配置
    ssh_configs[name] = new_config
    save_ssh_configs()
    
    # 如果该配置有活跃连接,先断开以便使用新配置
    client_key = f"{name}"
    if client_key in ssh_clients:
        ssh_disconnect(client_key)
    
    print(f"已更新SSH配置: {name}")
    return True

@mcp.tool(
    name='rename_ssh_config',
    description='重命名SSH连接配置'
)
def rename_ssh_config(old_name: str, new_name: str) -> bool:
    """
    重命名SSH连接配置
    
    Args:
        old_name: 当前配置名称
        new_name: 新的配置名称
    
    Returns:
        是否成功重命名配置
    """
    global ssh_configs, ssh_clients
    
    # 检查原配置是否存在
    if old_name not in ssh_configs:
        print(f"错误: 配置名称 '{old_name}' 不存在")
        return False
    
    # 检查新名称是否已被占用
    if new_name in ssh_configs:
        print(f"错误: 配置名称 '{new_name}' 已存在")
        return False
    
    # 获取原配置
    config = ssh_configs[old_name]
    
    # 如果该配置有活跃连接,先断开
    if old_name in ssh_clients:
        ssh_disconnect(old_name)
    
    # 使用新名称添加配置
    ssh_configs[new_name] = config
    
    # 删除旧配置
    del ssh_configs[old_name]
    
    # 保存更改
    save_ssh_configs()
    
    print(f"已将配置 '{old_name}' 重命名为 '{new_name}'")
    return True

@mcp.tool(
    name='list_ssh_configs',
    description='列出所有SSH连接配置'
)
def list_ssh_configs() -> Dict:
    """
    列出所有SSH连接配置
    
    Returns:
        所有SSH配置
    """
    configs_info = {}
    for name, config in ssh_configs.items():
        # 创建不包含密码的配置信息
        safe_config = config.copy()
        if "password" in safe_config:
            safe_config["password"] = "******" if safe_config["password"] else None
        configs_info[name] = safe_config
    
    return configs_info

@mcp.tool(
    name='ssh_connect',
    description='根据配置名称连接到SSH服务器'
)
def ssh_connect(config_name: str) -> str:
    """
    根据配置名称连接到SSH服务器
    
    Args:
        config_name: SSH配置名称
    
    Returns:
        连接成功返回客户端标识符,失败返回None
    """
    global ssh_clients, ssh_configs
    
    if config_name not in ssh_configs:
        print(f"错误: 未找到名为 '{config_name}' 的SSH配置")
        return None
    
    # 获取配置信息
    config = ssh_configs[config_name]
    hostname = config["hostname"]
    port = config["port"]
    username = config["username"]
    password = config["password"]
    key_filename = config["key_filename"]
    
    # 创建SSH客户端
    client = paramiko.SSHClient()
    
    # 自动添加策略,保存服务器的主机名和密钥信息
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    
    try:
        # 根据提供的认证方式连接SSH服务器
        if key_filename:
            # 使用SSH密钥认证
            client.connect(hostname=hostname, port=port, username=username, key_filename=key_filename)
            print(f"已成功使用SSH密钥连接到 {hostname}:{port}")
        else:
            # 使用密码认证
            client.connect(hostname=hostname, port=port, username=username, password=password)
            print(f"已成功使用密码连接到 {hostname}:{port}")
        
        # 使用配置名称作为连接标识符
        client_key = config_name
        ssh_clients[client_key] = client
        return client_key  # 返回连接的唯一标识符
    
    except Exception as e:
        print(f"连接失败: {str(e)}")
        return None
    
@mcp.tool(
    name='ssh_disconnect',
    description='关闭SSH连接'
)
def ssh_disconnect(client_key: str) -> bool:
    """
    关闭SSH连接
    
    Args:
        client_key: SSH连接的唯一标识符(配置名称)
    
    Returns:
        是否成功关闭连接
    """
    global ssh_clients
    client = ssh_clients.get(client_key)
    if not client:
        print(f"错误: SSH客户端 {client_key} 未连接")
        return False
    
    try:
        client.close()
        del ssh_clients[client_key]
        print(f"已关闭SSH连接: {client_key}")
        return True
    except Exception as e:
        print(f"关闭SSH连接失败: {str(e)}")
        return False

@mcp.tool(
    name='execute_ssh_command',
    description='在已连接的SSH客户端上执行命令并返回结果'
)
def execute_ssh_command(client_key: str, command: str) -> str:
    """
    在已连接的SSH客户端上执行命令并返回结果
    
    Args:
        client_key: SSH连接的唯一标识符(配置名称)
        command: 要执行的命令
    
    Returns:
        命令的输出结果
    """
    global ssh_clients
    client = ssh_clients.get(client_key)
    if not client:
        print(f"错误: SSH客户端 {client_key} 未连接")
        return None
    
    try:
        print(f"执行命令: {command}")
        
        # 执行命令
        stdin, stdout, stderr = client.exec_command(command)
        
        # 获取命令结果
        result = stdout.read().decode()
        error = stderr.read().decode()
        
        # 打印结果
        if result:
            print("命令输出:")
            print(result)
        
        # 打印错误(如果有)
        if error:
            print("错误信息:")
            print(error)
        
        return result
    
    except Exception as e:
        print(f"命令执行失败: {str(e)}")
        return None

if __name__ == "__main__":
    port = 9055
    app = mcp.sse_app()
    uvicorn.run(app, port=port, host='0.0.0.0')

结论

总而言之,MCP与自建工具调用流程在核心逻辑上并无本质区别。MCP通过引入客户端/服务端的架构和一套标准化协议(如代码示例所示的工具定义方式),为工具集成提供了一种规范化的选择。然而,其带来的直接技术优势相对于成熟的自建方案可能并不显著,尤其是在互操作性方面,并非不可替代。



网站公告

今日签到

点亮在社区的每一天
去签到