MCP (Model Context Protocol) 深入学习：构建 AI Agent 互联互通标准

前言

在 AI 应用蓬勃发展的今天，如何让大语言模型（LLM）与外部世界进行有效交互已成为开发者面临的核心挑战。传统的方式是针对每个应用场景编写定制化的 API 集成代码，这不仅导致大量重复工作，更难以实现不同 AI Agent 之间的互操作。正是在这一背景下，MCP（Model Context Protocol，模型上下文协议）应运而生。

MCP 是一种开放标准协议，旨在为 AI 模型提供一种统一的方式来发现和使用外部工具、数据源和功能。它的设计理念与 Web 时代的 REST API 类似——通过标准化接口实现系统间的互联互通。对于希望构建企业级 AI 应用或开发复杂 AI Agent 系统的中高级开发者而言，深入理解 MCP 已成为一项必备技能。

本文将带您系统学习 MCP 的核心概念，通过可运行的 Python 代码演示其使用方式，并结合实际场景探讨这一协议的应用价值。

核心概念

什么是 MCP？

MCP（Model Context Protocol）是一种基于 JSON-RPC 的通信协议，它定义了 AI 模型与外部系统之间的交互规范。MCP 的核心目标是将 AI 模型的"意图"转化为具体的外部操作，同时将外部世界的"状态"反馈给模型。

MCP 采用客户端-服务器架构，主要包含三个核心组件：

1. MCP Host（主机）：运行 AI 模型的应用程序，负责发起请求
2. MCP Client（客户端）：与 MCP Server 建立连接并转发请求
3. MCP Server（服务器）：提供工具（Tools）、资源（Resources）和提示（Prompts）

核心概念解析

工具（Tools）：MCP Server 向 AI 模型暴露的可执行功能。每个工具都有明确的输入输出模式（JSON Schema），AI 模型可以根据上下文智能选择调用哪些工具。

资源（Resources）：MCP Server 提供的数据内容，可以是文件、数据库记录、API 响应等。AI 模型可以读取这些资源来获取上下文信息。

提示（Prompts）：预定义的提示模板，可以被 AI 模型复用，帮助模型更好地理解特定场景。

采样（Sampling）：MCP 的一个强大特性——支持 Server 向 Client 请求让 LLM 生成内容，实现双向交互。

MCP 与 Function Calling 的区别

许多开发者可能会问：MCP 与 OpenAI 推出的 Function Calling 有什么区别？

特性	Function Calling	MCP
标准化程度	厂商特定	开放通用
状态管理	无状态	支持有状态会话
双向通信	单向调用	支持双向交互
工具发现	静态定义	动态发现
生态系统	封闭	开放生态

MCP 的设计更加面向复杂的企业应用场景，支持动态工具发现和会话状态管理，这使其成为构建大型 AI Agent 系统的理想选择。

代码示例

下面我们将通过 Python 代码演示如何实现一个基本的 MCP 客户端和服务器。代码基于 mcp 官方 Python SDK（假设已安装 mcp 包）：

# mcp_example.py
"""
MCP (Model Context Protocol) 实战示例
演示如何构建 MCP Server 和 Client
"""

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
from pydantic import AnyUrl
import asyncio
import json

# ============================================
# 第一部分：定义 MCP Server（提供工具和资源）
# ============================================

class MyMCPServer:
    """一个简单的 MCP 服务器实现"""
    
    def __init__(self):
        self.server = Server("my-mcp-server")
        self._register_handlers()
        # 模拟数据库
        self.user_data = {
            "users": [
                {"id": 1, "name": "张三", "role": "工程师"},
                {"id": 2, "name": "李四", "role": "产品经理"}
            ]
        }
    
    def _register_handlers(self):
        """注册工具处理器"""
        
        @self.server.list_tools()
        async def list_tools() -> list[Tool]:
            """列出所有可用的工具"""
            return [
                Tool(
                    name="get_user_by_id",
                    description="根据用户ID获取用户信息",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "user_id": {
                                "type": "integer",
                                "description": "用户ID"
                            }
                        },
                        "required": ["user_id"]
                    }
                ),
                Tool(
                    name="create_user",
                    description="创建新用户",
                    inputSchema={
                        "type": "object",
                        "properties": {
                            "name": {"type": "string", "description": "用户名"},
                            "role": {"type": "string", "description": "用户角色"}
                        },
                        "required": ["name", "role"]
                    }
                ),
                Tool(
                    name="list_users",
                    description="列出所有用户",
                    inputSchema={
                        "type": "object",
                        "properties": {}
                    }
                )
            ]
        
        @self.server.call_tool()
        async def call_tool(name: str, arguments: dict) -> list[TextContent]:
            """处理工具调用请求"""
            
            if name == "get_user_by_id":
                user_id = arguments.get("user_id")
                user = next(
                    (u for u in self.user_data["users"] if u["id"] == user_id),
                    None
                )
                if user:
                    return [TextContent(type="text", text=json.dumps(user, ensure_ascii=False))]
                return [TextContent(type="text", text="用户不存在")]
            
            elif name == "create_user":
                new_id = max(u["id"] for u in self.user_data["users"]) + 1
                new_user = {
                    "id": new_id,
                    "name": arguments.get("name"),
                    "role": arguments.get("role")
                }
                self.user_data["users"].append(new_user)
                return [TextContent(type="text", text=json.dumps(new_user, ensure_ascii=False))]
            
            elif name == "list_users":
                return [TextContent(
                    type="text",
                    text=json.dumps(self.user_data["users"], ensure_ascii=False)
                )]
            
            return [TextContent(type="text", text="未知工具")]
    
    async def run(self):
        """运行 MCP 服务器"""
        async with stdio_server() as (read_stream, write_stream):
            await self.server.run(
                read_stream,
                write_stream,
                self.server.create_initialization_options()
            )


# ============================================
# 第二部分：定义 MCP Client（调用工具）
# ============================================

class MCPClient:
    """MCP 客户端实现"""
    
    def __init__(self, server_command: list):
        import subprocess
        self.process = subprocess.Popen(
            server_command,
            stdout=subprocess.PIPE,
            stdin=subprocess.PIPE
        )
    
    async def call_tool(self, tool_name: str, arguments: dict):
        """调用远程工具"""
        # 构建 JSON-RPC 请求
        request = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {
                "name": tool_name,
                "arguments": arguments
            }
        }
        
        # 发送请求并接收响应（简化版）
        # 实际使用时应使用 mcp.Client 类的完整实现
        print(f"调用工具: {tool_name}")
        print(f"参数: {arguments}")
        
        return {"status": "success", "tool": tool_name}


# ============================================
# 第三部分：模拟 LLM 与 MCP 的集成
# ============================================

class MockLLMWithMCP:
    """
    模拟 LLM 与 MCP 工具调用的集成
    实际生产环境中可替换为 OpenAI、Claude 等真实 LLM
    """
    
    def __init__(self, mcp_client: MCPClient):
        self.mcp_client = mcp_client
        # 工具描述（LLM 可以理解这些工具的用途）
        self.tools_description = """
可用工具：
1. get_user_by_id(user_id: int) -> 获取指定ID的用户信息
2. create_user(name: str, role: str) -> 创建新用户
3. list_users() -> 获取所有用户列表
"""
    
    async def process_request(self, user_prompt: str):
        """
        处理用户请求并决定是否调用工具
        这里是简化版的意图识别和工具调用逻辑
        """
        # 模拟 LLM 的决策过程
        # 实际应用中，这里会调用真实的 LLM API
        print(f"收到请求: {user_prompt}")
        
        # 简单的规则匹配（生产环境应使用 LLM）
        if "列出" in user_prompt or "所有用户" in user_prompt:
            result = await self.mcp_client.call_tool("list_users", {})
            return f"当前所有用户：{result}"
        
        elif "创建" in user_prompt and "用户" in user_prompt:
            # 简化参数提取
            name = "新用户"
            role = "默认角色"
            result = await self.mcp_client.call_tool("create_user", {
                "name": name,
                "role": role
            })
            return f"用户创建成功：{result}"
        
        return "无法理解请求"


# ============================================
# 主函数：启动服务器
# ============================================

async def main():
    server = MyMCPServer()
    print("MCP 服务器启动中...")
    await server.run()

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

以上代码展示了一个完整的 MCP Server 实现，包括工具定义、参数校验和响应处理。在实际项目中，你可以根据需要扩展更多工具和功能。

实战场景

场景一：智能客服助手

在电商平台的客服场景中，系统需要处理用户的查询并决定是否需要调用后端API获取订单信息、仓库数据或物流状态。当用户询问"我的订单到哪里了"时，Agent解析意图后发现需要查询物流信息，于是调用订单追踪工具获取最新物流动态，再将结果自然语言化后回复用户。

# 示例：智能客服Agent处理物流查询
class CustomerServiceAgent:
    def __init__(self, tools):
        self.llm = get_llm()
        self.tools = tools
    
    def process_request(self, user_message):
        # 1. 解析用户意图并决定是否调用工具
        intent = self.classify_intent(user_message)
        
        if intent == "query_order_status":
            # 2. 调用物流查询工具
            order_id = self.extract_order_id(user_message)
            result = self.tools["query_logistics"](order_id)
            return self.format_response(result)
        
        elif intent == "general_question":
            # 3. 直接使用LLM回答
            return self.llm.chat(user_message)
        
        else:
            return "抱歉，我无法理解您的问题，请重述。"

场景二：代码审查助手

在软件开发流程中，AI Agent可以辅助进行代码审查。当开发者提交代码变更时，Agent自动分析Diff内容，识别潜在的安全风险、代码异味或性能问题，并根据严重程度决定是否调用静态分析工具深入检查，或直接给出修复建议。

场景三：自动化数据分析

在数据分析场景中，Agent接收用户的自然语言查询（如"分析上季度的销售趋势"），自动解析需求并调用相应的数据处理工具。在复杂场景下，Agent还能进行多轮工具调用，先提取数据、再进行清洗、最后生成可视化报表，真正实现端到端的智能分析体验。

---

总结

构建高效的AI Agent系统，核心在于打造一个智能的"大脑"来协调用户的请求与工具的执行。通过精心设计的Prompt工程，我们可以让LLM具备意图识别、参数提取和结果解读的完整能力；通过定义清晰的工具接口和返回值，Agent能够准确调用外部服务并把结果转化为用户可以理解的自然语言。

整个系统运转的关键流程包括：

• 接收请求：解析用户的自然语言输入
• 意图判断：基于Prompt Engineering让LLM决定是否需要调用工具
• 参数提取：从用户输入中提取调用工具所需的参数
• 工具执行：调用相应的外部API或服务
• 结果生成：将工具返回的原始数据转化为自然语言回复

Python凭借其丰富的生态和简洁的语法，为AI Agent的开发提供了理想的土壤。随着LLM API的持续进化和开源工具的日益丰富，基于Python构建智能Agent将成为AI应用的主流范式。

---

标签：python, 人工智能, ai