第10章：模型上下文协议

为了使LLM能够有效地作为智能体运行，它们的能力必须超越多模态生成。与外部环境的交互是必要的，包括访问当前数据、利用外部软件以及执行特定的操作任务。模型上下文协议（MCP）通过为LLM与外部资源接口提供标准化接口来解决这一需求。此协议作为促进一致和可预测集成的关键机制。

MCP模式概述

想象一个通用适配器，允许任何LLM插入任何外部系统、数据库或工具，而无需为每个系统进行自定义集成。这本质上就是模型上下文协议（MCP）的作用。它是一个开放标准，旨在标准化像Gemini、OpenAI的GPT模型、Mixtral和Claude这样的LLM与外部应用程序、数据源和工具的通信方式。将其视为简化LLM获取上下文、执行行动并与各种系统交互的通用连接机制。

MCP在客户端-服务器架构上运行。它定义了MCP服务器如何暴露不同元素——数据（称为资源）、交互模板（本质上是提示）和可操作函数（称为工具）。然后这些被MCP客户端消费，MCP客户端可能是LLM主机应用程序或AI智能体本身。这种标准化方法大大减少了将LLM集成到不同操作环境中的复杂性。

然而，MCP是"智能体接口"的合同，其有效性在很大程度上取决于它暴露的底层API的设计。存在开发者简单地包装预先存在的、遗留API而不进行修改的风险，这对智能体来说可能是次优的。例如，如果票务系统的API只允许一次检索一个完整票务详情，被要求总结高优先级票务的智能体在高容量下会缓慢且不准确。要真正有效，底层API应该通过过滤和排序等确定性功能进行改进，以帮助非确定性智能体高效工作。这突出了智能体不会神奇地替换确定性工作流；它们通常需要更强的确定性支持才能成功。

此外，MCP可以包装输入或输出仍然不是智能体固有可理解的API。只有当其数据格式对智能体友好时，API才有用，这是MCP本身不强制执行的保证。例如，为返回PDF文件的文档存储创建MCP服务器，如果消费智能体无法解析PDF内容，则大多无用。更好的方法是首先创建一个返回文档文本版本（如Markdown）的API，智能体实际上可以读取和处理。这演示了开发者必须考虑不仅仅是连接，还要考虑交换数据的性质以确保真正的兼容性。

MCP与工具函数调用

模型上下文协议（MCP）和工具函数调用是使LLM能够与外部能力（包括工具）交互并执行行动的不同机制。虽然两者都用于扩展LLM超越文本生成的能力，但它们在方法和抽象级别上有所不同。

工具函数调用可以被视为LLM对特定、预定义工具或函数的直接请求。注意，在此上下文中，我们互换使用"工具"和"函数"这两个词。这种交互的特征是一对一通信模型，其中LLM基于其对需要外部行动的用户意图的理解格式化请求。然后应用程序代码执行此请求并将结果返回给LLM。此过程通常是专有的，在不同的LLM提供商之间有所不同。

相比之下，模型上下文协议（MCP）作为LLM发现、通信和利用外部能力的标准化接口运行。它作为促进与广泛工具和系统交互的开放协议，旨在建立一个生态系统，其中任何兼容的工具都可以被任何兼容的LLM访问。这促进了不同系统和实现之间的互操作性、可组合性和可重用性。通过采用联合模型，我们显著提高了互操作性并释放了现有资产的价值。这种策略允许我们通过简单地将它们包装在MCP兼容接口中，将不同的和遗留服务带入现代生态系统。这些服务继续独立运行，但现在可以组合成新的应用程序和工作流，它们的协作由LLM编排。这促进了敏捷性和可重用性，而无需对基础系统进行昂贵的重写。

以下是MCP和工具函数调用之间基本区别的分解：

特性	工具函数调用	模型上下文协议（MCP）
标准化	专有且特定于供应商。格式和实现在LLM提供商之间不同。	开放、标准化协议，促进不同LLM和工具之间的互操作性。
范围	LLM请求执行特定、预定义函数的直接机制。	LLM和外部工具如何发现和相互通信的更广泛框架。
架构	LLM与应用程序工具处理逻辑之间的一对一交互。	客户端-服务器架构，其中LLM驱动的应用程序（客户端）可以连接并利用各种MCP服务器（工具）。
发现	LLM被明确告知在特定对话上下文中哪些工具可用。	启用可用工具的动态发现。MCP客户端可以查询服务器以查看它提供什么能力。
可重用性	工具集成通常与使用的特定应用程序和LLM紧密耦合。	促进可重用、独立的"MCP服务器"的开发，任何兼容的应用程序都可以访问。

将工具函数调用视为给AI一套特定的定制工具，如特定的扳手和螺丝刀。这对于具有固定任务集的车间是有效的。另一方面，MCP（模型上下文协议）就像创建一个通用、标准化的电源插座系统。它本身不提供工具，但它允许任何制造商的任何兼容工具插入并工作，实现动态且不断扩展的车间。

简而言之，函数调用提供对几个特定函数的直接访问，而MCP是让LLM发现和使用大量外部资源的标准化通信框架。对于简单应用程序，特定工具就足够了；对于需要适应的复杂、互连的AI系统，像MCP这样的通用标准是必不可少的。

MCP的额外考虑

虽然MCP呈现了一个强大的框架，但彻底评估需要考虑影响其适合给定用例的几个关键方面。让我们更详细地看看一些方面：

工具与资源与提示：理解这些组件的特定角色很重要。资源是静态数据（如PDF文件、数据库记录）。工具是执行行动的可执行函数（如发送电子邮件、查询API）。提示是指导LLM如何与资源或工具交互的模板，确保交互是结构化和有效的。
可发现性：MCP的一个关键优势是MCP客户端可以动态查询服务器以了解它提供什么工具和资源。这种"即时"发现机制对于需要适应新能力而无需重新部署的智能体来说是强大的。
安全性：通过任何协议暴露工具和数据需要强大的安全措施。MCP实现必须包括身份验证和授权，以控制哪些客户端可以访问哪些服务器以及允许它们执行什么特定行动。
实现：虽然MCP是开放标准，但其实现可能复杂。然而，提供商开始简化此过程。例如，一些模型提供商如Anthropic或FastMCP提供SDK，抽象了大部分样板代码，使开发者更容易创建和连接MCP客户端和服务器。
错误处理：全面的错误处理策略至关重要。协议必须定义如何将错误（如工具执行失败、服务器不可用、无效请求）传达回LLM，以便它能够理解失败并可能尝试替代方法。

实践代码示例

MCP客户端实现

import asyncio
import json
from typing import Dict, Any, List
import aiohttp

class MCPClient:
    """MCP客户端实现"""
    
    def __init__(self, server_url: str, api_key: str = None):
        self.server_url = server_url
        self.api_key = api_key
        self.session = None
        
    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self
        
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()
            
    async def discover_capabilities(self) -> Dict[str, Any]:
        """发现服务器能力"""
        headers = {"Content-Type": "application/json"}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"
            
        async with self.session.get(
            f"{self.server_url}/capabilities",
            headers=headers
        ) as response:
            if response.status == 200:
                return await response.json()
            else:
                raise Exception(f"发现能力失败: {response.status}")
                
    async def list_resources(self) -> List[Dict[str, Any]]:
        """列出可用资源"""
        headers = {"Content-Type": "application/json"}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"
            
        async with self.session.get(
            f"{self.server_url}/resources",
            headers=headers
        ) as response:
            if response.status == 200:
                return await response.json()
            else:
                raise Exception(f"列出资源失败: {response.status}")
                
    async def get_resource(self, resource_id: str) -> Dict[str, Any]:
        """获取特定资源"""
        headers = {"Content-Type": "application/json"}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"
            
        async with self.session.get(
            f"{self.server_url}/resources/{resource_id}",
            headers=headers
        ) as response:
            if response.status == 200:
                return await response.json()
            else:
                raise Exception(f"获取资源失败: {response.status}")
                
    async def list_tools(self) -> List[Dict[str, Any]]:
        """列出可用工具"""
        headers = {"Content-Type": "application/json"}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"
            
        async with self.session.get(
            f"{self.server_url}/tools",
            headers=headers
        ) as response:
            if response.status == 200:
                return await response.json()
            else:
                raise Exception(f"列出工具失败: {response.status}")
                
    async def call_tool(self, tool_name: str, parameters: Dict[str, Any]) -> Dict[str, Any]:
        """调用工具"""
        headers = {"Content-Type": "application/json"}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"
            
        payload = {
            "tool": tool_name,
            "parameters": parameters
        }
        
        async with self.session.post(
            f"{self.server_url}/tools/call",
            headers=headers,
            json=payload
        ) as response:
            if response.status == 200:
                return await response.json()
            else:
                raise Exception(f"调用工具失败: {response.status}")

# 使用示例
async def use_mcp_client():
    """使用MCP客户端"""
    async with MCPClient("http://localhost:8000", "your-api-key") as client:
        # 发现能力
        capabilities = await client.discover_capabilities()
        print(f"服务器能力: {capabilities}")
        
        # 列出资源
        resources = await client.list_resources()
        print(f"可用资源: {resources}")
        
        # 列出工具
        tools = await client.list_tools()
        print(f"可用工具: {tools}")
        
        # 调用工具
        if tools:
            tool_name = tools[0]["name"]
            result = await client.call_tool(tool_name, {"param1": "value1"})
            print(f"工具调用结果: {result}")

if __name__ == "__main__":
    asyncio.run(use_mcp_client())

MCP服务器实现

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Dict, Any, List
import uvicorn

app = FastAPI(title="MCP Server", version="1.0.0")

# 数据模型
class ToolCall(BaseModel):
    tool: str
    parameters: Dict[str, Any]

class Resource(BaseModel):
    id: str
    name: str
    description: str
    type: str
    content: Any

class Tool(BaseModel):
    name: str
    description: str
    parameters: Dict[str, Any]

# 模拟数据存储
resources_db = {
    "doc1": Resource(
        id="doc1",
        name="用户手册",
        description="产品用户手册",
        type="document",
        content="这是用户手册的内容..."
    ),
    "data1": Resource(
        id="data1",
        name="销售数据",
        description="2024年销售数据",
        type="data",
        content={"sales": 1000, "revenue": 50000}
    )
}

tools_db = {
    "search": Tool(
        name="search",
        description="搜索资源",
        parameters={"query": "string", "type": "string"}
    ),
    "analyze": Tool(
        name="analyze",
        description="分析数据",
        parameters={"data_id": "string", "analysis_type": "string"}
    )
}

@app.get("/capabilities")
async def get_capabilities():
    """获取服务器能力"""
    return {
        "version": "1.0.0",
        "capabilities": {
            "resources": True,
            "tools": True,
            "prompts": False
        }
    }

@app.get("/resources")
async def list_resources():
    """列出所有资源"""
    return list(resources_db.values())

@app.get("/resources/{resource_id}")
async def get_resource(resource_id: str):
    """获取特定资源"""
    if resource_id not in resources_db:
        raise HTTPException(status_code=404, detail="资源未找到")
    return resources_db[resource_id]

@app.get("/tools")
async def list_tools():
    """列出所有工具"""
    return list(tools_db.values())

@app.post("/tools/call")
async def call_tool(tool_call: ToolCall):
    """调用工具"""
    if tool_call.tool not in tools_db:
        raise HTTPException(status_code=404, detail="工具未找到")
    
    tool = tools_db[tool_call.tool]
    
    # 模拟工具执行
    if tool_call.tool == "search":
        query = tool_call.parameters.get("query", "")
        resource_type = tool_call.parameters.get("type", "")
        
        results = []
        for resource in resources_db.values():
            if query.lower() in resource.name.lower() or query.lower() in resource.description.lower():
                if not resource_type or resource.type == resource_type:
                    results.append(resource)
        
        return {"results": results, "count": len(results)}
    
    elif tool_call.tool == "analyze":
        data_id = tool_call.parameters.get("data_id", "")
        analysis_type = tool_call.parameters.get("analysis_type", "basic")
        
        if data_id not in resources_db:
            raise HTTPException(status_code=404, detail="数据资源未找到")
        
        resource = resources_db[data_id]
        
        if analysis_type == "basic":
            return {"analysis": f"对{resource.name}的基本分析完成"}
        elif analysis_type == "detailed":
            return {"analysis": f"对{resource.name}的详细分析完成"}
        else:
            return {"analysis": f"对{resource.name}的{analysis_type}分析完成"}
    
    return {"message": "工具执行完成"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

集成示例

from langchain.agents import Tool, AgentExecutor, create_react_agent
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
import asyncio

class MCPIntegratedAgent:
    """集成MCP的智能体"""
    
    def __init__(self, mcp_client: MCPClient):
        self.mcp_client = mcp_client
        self.llm = ChatOpenAI(model="gpt-4", temperature=0)
        self.tools = []
        
    async def initialize(self):
        """初始化智能体"""
        # 发现MCP服务器能力
        capabilities = await self.mcp_client.discover_capabilities()
        print(f"发现能力: {capabilities}")
        
        # 获取可用工具
        mcp_tools = await self.mcp_client.list_tools()
        
        # 将MCP工具转换为LangChain工具
        for tool in mcp_tools:
            langchain_tool = Tool(
                name=tool["name"],
                description=tool["description"],
                func=self.create_tool_function(tool["name"])
            )
            self.tools.append(langchain_tool)
            
    def create_tool_function(self, tool_name: str):
        """为MCP工具创建LangChain工具函数"""
        async def tool_function(parameters: str) -> str:
            try:
                # 解析参数
                params = json.loads(parameters) if parameters else {}
                
                # 调用MCP工具
                result = await self.mcp_client.call_tool(tool_name, params)
                return json.dumps(result, ensure_ascii=False)
            except Exception as e:
                return f"工具调用失败: {str(e)}"
                
        return tool_function
        
    async def run(self, query: str):
        """运行智能体"""
        # 创建提示模板
        prompt = PromptTemplate(
            template="""
            你是一个智能助手，可以访问外部工具和资源。
            
            可用工具:
            {tools}
            
            工具名称: {tool_names}
            
            用户问题: {input}
            
            请使用适当的工具来回答用户的问题。
            """,
            input_variables=["input", "tools", "tool_names"]
        )
        
        # 创建智能体
        agent = create_react_agent(self.llm, self.tools, prompt)
        agent_executor = AgentExecutor(agent=agent, tools=self.tools, verbose=True)
        
        # 运行智能体
        result = agent_executor.invoke({"input": query})
        return result

# 使用示例
async def run_mcp_agent():
    """运行MCP集成智能体"""
    async with MCPClient("http://localhost:8000", "your-api-key") as client:
        agent = MCPIntegratedAgent(client)
        await agent.initialize()
        
        # 运行智能体
        result = await agent.run("请搜索用户手册并分析销售数据")
        print(f"智能体结果: {result}")

if __name__ == "__main__":
    asyncio.run(run_mcp_agent())