乐于分享
好东西不私藏

【AI智能体设计模式】第10章 模型上下文协议(MCP,Multi Context Protocol)

【AI智能体设计模式】第10章 模型上下文协议(MCP,Multi Context Protocol)

MCP 模式概述

模型上下文协议(Model Context Protocol,MCP)是一种标准化的协议,用于在 AI 模型和外部数据源、工具之间建立连接。它提供了一种统一的方式来描述、发现和使用外部资源,使智能体能够访问各种数据源和工具,而不需要为每个资源单独集成。

MCP 的核心思想是将"上下文"作为一等公民,通过标准化协议管理上下文的获取、转换和使用。

MCP 与工具函数调用的区别

工具函数调用

  • 智能体主动调用特定函数
  • 需要预先定义每个工具
  • 工具是被动等待调用的
  • 适合明确的、单一的操作

MCP

  • 提供标准化的上下文获取机制
  • 资源可以主动提供上下文
  • 支持动态发现和使用资源
  • 适合复杂的、多源的上下文管理

MCP 的更多考量

优势

  • 标准化
    :统一的协议简化集成
  • 可扩展
    :轻松添加新资源
  • 灵活性
    :支持多种上下文类型
  • 安全性
    :内置权限和访问控制

挑战

  • 复杂度
    需要理解和实现协议
  • 性能
    额外的协议层可能影响性能
  • 兼容性
    需要确保资源兼容性

实践应用与场景

1. 多数据源集成

从多个数据库、API 和文件系统获取数据,统一提供给智能体。

2. 企业知识管理

连接企业内部知识库、文档系统和数据库,构建企业级智能助手。

3. 开发工具集成

集成 IDE、版本控制、项目管理等工具,构建开发助手。

4. 实时数据访问

连接实时数据源(如市场数据、传感器数据),支持实时决策。

MCP 中的设计模式应用

MCP 的实现充分利用了多种经典设计模式,这些模式共同构成了 MCP 灵活、可扩展的架构基础。

1. 装饰器模式

装饰器模式用于动态地给方法添加额外功能。在 MCP 中,@Resource 和 @Tool 装饰器将普通方法转换为资源或工具,为其添加元数据和注册功能。

应用场景:将普通方法标记为 MCP 资源或工具,无需修改方法本身。

代码示例

from functools import wrapsdef resource(uri: str):    """资源装饰器:将方法标记为MCP资源"""    def decorator(func):        @wraps(func)        async def wrapper(self, *args, **kwargs):            wrapper._is_resource = True            wrapper._uri = uri            result = await func(self, *args, **kwargs)            return result        wrapper._is_resource = True        wrapper._uri = uri        return wrapper    return decoratordef tool(name: str):    """工具装饰器:将方法标记为MCP工具"""    def decorator(func):        @wraps(func)        async def wrapper(self, *args, **kwargs):            wrapper._is_tool = True            wrapper._tool_name = name            result = await func(self, *args, **kwargs)            return result        wrapper._is_tool = True        wrapper._tool_name = name        return wrapper    return decoratorclass MyMCPServer:    @resource("database://users")    async def get_users(self, query: str):        return {"users": []}    @tool("search")    async def search(self, query: str):        return {"results": []}

2. 模板方法模式

模板方法模式在基类中定义算法骨架,由子类实现具体步骤。MCPServer 基类定义了服务器的标准流程,子类通过实现具体方法来定制行为。

应用场景:定义 MCP 服务器的标准生命周期和操作流程。

代码示例

from abc import ABC, abstractmethodclass MCPServer(ABC):    """MCP服务器基类,定义模板方法"""    async def start(self):        """启动服务器的模板方法"""        await self.initialize()        await self.register_resources()        await self.register_tools()        await self.run()    async def initialize(self):        """初始化服务器"""        print(f"服务器 {self.name} 初始化中...")    @abstractmethod    async def register_resources(self):        """注册资源(子类实现)"""        pass    @abstractmethod    async def register_tools(self):        """注册工具(子类实现)"""        pass    async def run(self):        """运行服务器"""        print("服务器运行中...")class MyMCPServer(MCPServer):    def __init__(self, name: str):        self.name = name    async def register_resources(self):        print("注册用户资源...")    async def register_tools(self):        print("注册搜索工具...")

3. 外观模式

外观模式为复杂子系统提供简单接口。MCPClient 封装了连接管理、资源发现、工具调用等复杂操作,为用户提供统一简单的 API。

应用场景:简化 MCP 协议的复杂操作流程。

代码示例

class MCPClient:    """MCP客户端外观,提供简化的操作接口"""    def __init__(self):        self._connection = None        self._resource_cache = {}        self._tool_registry = {}    async def connect(self, server_name: str):        """连接到MCP服务器"""        self._connection = await self._establish_connection(server_name)        await self._discover_resources()        await self._discover_tools()        print(f"已连接到 {server_name}")    async def list_resources(self):        """列出所有可用资源"""        return list(self._resource_cache.keys())    async def get_resource(self, uri: str, **params):        """获取资源数据"""        resource = self._resource_cache.get(uri)        if resource:            return await resource.fetch(**params)        raise ValueError(f"资源 {uri} 不存在")    async def call_tool(self, name: str, **params):        """调用工具"""        tool = self._tool_registry.get(name)        if tool:            return await tool.execute(**params)        raise ValueError(f"工具 {name} 不存在")    async def _establish_connection(self, server_name: str):        return {"server": server_name}    async def _discover_resources(self):        self._resource_cache = {"database://users": Resource()}    async def _discover_tools(self):        self._tool_registry = {"search": Tool()}# 用户使用简化的接口client = MCPClient()await client.connect("my_server")data = await client.get_resource("database://users")

4. 策略模式

策略模式定义一系列算法,让它们可以相互替换。MCP 支持不同的资源获取策略(数据库、文件、API等),通过统一接口调用。

应用场景:支持多种数据源类型,每种类型有不同的获取策略。

代码示例

from abc import ABC, abstractmethodfrom typing import Anyclass ResourceStrategy(ABC):    """资源获取策略接口"""    @abstractmethod    async def fetch(self, query: str) -> Any:        passclass DatabaseStrategy(ResourceStrategy):    """数据库资源获取策略"""    async def fetch(self, query: str) -> Any:        print(f"从数据库查询: {query}")        return {"source": "database", "data": []}class FileStrategy(ResourceStrategy):    """文件资源获取策略"""    async def fetch(self, query: str) -> Any:        print(f"从文件读取: {query}")        return {"source": "file", "data": []}class APIStrategy(ResourceStrategy):    """API资源获取策略"""    async def fetch(self, query: str) -> Any:        print(f"从API调用: {query}")        return {"source": "api", "data": []}class ResourceContext:    """资源上下文,使用策略"""    def __init__(self, strategy: ResourceStrategy):        self._strategy = strategy    async def get_data(self, query: str):        return await self._strategy.fetch(query)# 使用不同的策略db_resource = ResourceContext(DatabaseStrategy())file_resource = ResourceContext(FileStrategy())api_resource = ResourceContext(APIStrategy())data1 = await db_resource.get_data("SELECT * FROM users")data2 = await file_resource.get_data("/data/users.json")data3 = await api_resource.get_data("https://api.example.com/users")

5. 工厂方法模式

工厂方法模式让子类决定创建哪种对象。MCP 中的装饰器实际上充当工厂,根据装饰器参数创建不同类型的 Resource 或 Tool 对象。

应用场景:根据配置动态创建不同类型的资源或工具对象。

代码示例

from abc import ABC, abstractmethodclass Resource(ABC):    """资源抽象产品"""    @abstractmethod    async def fetch(self, **params):        passclass DatabaseResource(Resource):    """数据库资源具体产品"""    def __init__(self, uri: str):        self.uri = uri    async def fetch(self, **params):        print(f"从数据库 {self.uri} 获取数据")        return {"type": "database", "uri": self.uri}class FileResource(Resource):    """文件资源具体产品"""    def __init__(self, uri: str):        self.uri = uri    async def fetch(self, **params):        print(f"从文件 {self.uri} 读取数据")        return {"type": "file", "uri": self.uri}class ResourceFactory:    """资源工厂"""    @staticmethod    def create_resource(uri: str) -> Resource:        """根据URI创建对应的资源对象"""        if uri.startswith("database://"):            return DatabaseResource(uri)        elif uri.startswith("file://"):            return FileResource(uri)        else:            raise ValueError(f"不支持的资源类型: {uri}")# 使用工厂创建资源db_resource = ResourceFactory.create_resource("database://users")file_resource = ResourceFactory.create_resource("file:///data/config.json")data1 = await db_resource.fetch()data2 = await file_resource.fetch()

设计模式协同工作

这些设计模式在 MCP 中协同工作,共同构建了一个灵活、可扩展的系统:

  • 装饰器模式
    为方法添加元数据,实现声明式编程
  • 模板方法模式
    定义服务器生命周期,确保一致性
  • 外观模式
    简化客户端使用,隐藏复杂细节
  • 策略模式
    支持多种资源类型,易于扩展
  • 工厂方法模式
    动态创建对象,提高灵活性

这种多模式组合体现了 MCP 的设计哲学:通过标准化协议和经典模式,构建可维护、可扩展的智能体资源集成架构。

一图速览

是什么

智能体需要访问多种外部资源,但缺乏统一的集成方式,导致每个资源都需要单独开发,集成复杂,维护困难。

为什么

MCP 通过标准化协议,提供统一的资源发现、访问和使用机制,简化集成,提高可扩展性和可维护性。

经验法则

当智能体需要访问多种外部资源,或需要构建可扩展的资源集成架构时,应采用 MCP。

关键要点

  • MCP 提供标准化的上下文管理协议。
  • 区别于工具调用,MCP 更强调上下文的主动提供和动态发现。
  • 支持多数据源集成、企业知识管理等场景。
  • Google ADK 提供了 MCP 的完整实现。
  • MCP 是构建企业级、可扩展智能体的重要工具。

总结

模型上下文协议(MCP)为智能体与外部资源的集成提供了标准化解决方案。通过统一协议管理上下文,MCP 简化了集成复杂度,提高了系统的可扩展性和可维护性。对于需要访问多种数据源和工具的智能体系统,MCP 是重要的架构选择。

参考资料

  • Model Context Protocol 规范 - modelcontextprotocol.io
  • Google ADK MCP 文档 - google.github.io
  • MCP 实现示例 - github.com

参照书籍《Agentic Design Patterns》的基本概念和观点。