MCP协议详细教程

@小森

发布于 2025-07-26 08:01:01

26500

代码可运行

文章被收录于专栏：xiaosenxiaosen

运行总次数：0

代码可运行

构建智能代理：MCP协议实现AI与外部资源的无缝集成

引言

Model Context Protocol (MCP) 是一个开放协议，旨在标准化应用程序为大语言模型(LLM)提供上下文的方式。可以将MCP比作AI应用的USB-C接口——正如USB-C为设备连接各种外设提供了标准化方式，MCP为AI模型连接不同数据源和工具提供了标准化方式。

本教程将指导您在AI代理应用中实现MCP，演示如何通过提供对外部资源、工具和数据源的无缝访问来增强代理的能力。

为什么MCP对代理如此重要

传统的AI模型与外部资源连接方法通常需要为每个数据源或工具进行自定义集成，这导致了：

集成复杂性：每个新数据源都需要独特的实现
可扩展性问题：添加新工具变得越来越困难
维护开销：对一个集成的更新可能会破坏其他集成

MCP通过提供标准化协议解决了这些挑战，实现了：

统一访问：多个数据源和工具的单一接口
即插即用扩展：轻松添加新功能
有状态通信：AI与资源之间的实时双向通信
动态发现：AI可以即时发现和使用新工具

官方MCP服务器示例

MCP社区维护着一系列参考服务器实现集合，展示了最佳实践并演示了各种集成模式。这些官方示例可在 MCP Servers 获得，为希望创建自己MCP服务器的开发者提供了宝贵的起点。

我们将构建什么

在本教程中，我们将实现：

构建并使用您的MCP服务器：构建具有自定义工具的MCP服务器并连接到Claude Desktop
自定义工具启用代理：创建一个可以通过MCP使用外部工具的自定义代理

完成本教程后，您将理解MCP如何通过提供对更广泛数字生态系统的访问来增强您的AI代理，使它们更有能力、更具上下文感知能力和更有用。

MCP架构概述

MCP遵循客户端-服务器架构，包含三个主要组件：

Host：需要访问外部资源的AI应用（如Claude Desktop、Cursor或自定义代理）
Clients：与服务器保持连接的连接器
Servers：通过MCP协议公开功能（数据、工具、提示）的轻量级程序
Data Sources：MCP服务器可以访问的本地（文件、数据库）和远程服务（API）

MCP内的通信使用基于WebSocket连接的JSON-RPC 2.0，确保组件之间的实时双向通信。

体验MCP：先试用再构建

虽然本教程专注于构建自己的MCP服务器并将其与AI代理集成，但您可能希望在深入开发之前快速体验MCP在实践中的工作原理。

官方MCP文档为希望在Claude Desktop或其他兼容AI应用中尝试现有MCP服务器的用户提供了出色的快速入门指南。这让您无需编写任何代码就能亲身感受MCP启用的功能。

👉 亲自尝试： MCP用户快速入门指南

通过探索快速入门指南，您将获得对我们在本教程中构建内容的实用洞察。当您准备好理解内部工作原理并创建自己的实现时，请继续我们下面的逐步开发过程。

现在，让我们开始构建自己的MCP服务器和客户端！

构建您的MCP服务器

现在我们了解了MCP的基础知识，让我们构建第一个MCP服务器！在本节中，我们将使用CoinGecko API创建一个加密货币价格查询服务。我们的服务器将提供允许AI检查加密货币当前价格或市场数据的工具。

设置环境

在深入实现之前，让我们安装必要的包并设置环境。

注意： 对于安装步骤，请打开终端窗口。这些命令应在常规终端中运行，而不是在Jupyter notebook单元格中。

步骤1：安装uv包管理器

# 在终端中运行，不在Jupyter中
curl -LsSf https://astral.sh/uv/install.sh | sh

步骤2：设置项目

# 创建并导航到项目目录
mkdir mcp-crypto-server
cd mcp-crypto-server
uv init

# 创建并激活虚拟环境
uv venv
source .venv/bin/activate  # Windows上：.venv\Scripts\activate

# 安装依赖项
uv add "mcp[cli]" httpx

运行MCP服务器

设置环境后，我们可以开始构建工具。

请查看 mcp_server.py 了解如何构建工具。

现在我们可以通过在终端中运行以下命令来启动服务器：

# 从scripts文件夹复制服务器文件
cp ../scripts/mcp_server.py .

# 启动MCP服务器
uv run mcp_server.py

与Claude Desktop集成

如果您还没有下载Claude Desktop，请查看此页面。

要将您的MCP服务器连接到Claude Desktop：

步骤1：查找uv命令的绝对路径：

which uv

复制输出（例如，/user/local/bin/uv或类似路径）

步骤2：创建或编辑Claude Desktop配置文件：

macOS：~/Library/Application Support/Claude/claude_desktop_config.json
Windows：%APPDATA%\Claude\claude_desktop_config.json
Linux：~/.config/Claude/claude_desktop_config.json

您可以查看此页面了解如何创建配置文件。

步骤3：添加您的MCP服务器配置：

{
    "mcpServers": {
        "crypto-price-tracker": {
            "command": "/ABSOLUTE/PATH/TO/uv",
            "args": [
                "--directory",
                "/ABSOLUTE/PATH/TO/GenAI_Agents/all_agents_tutorials/mcp-crypto-server",
                "run",
                "mcp_server.py"
            ]
        }
    }
}

将/ABSOLUTE/PATH/TO/uv替换为您从which uv命令获得的路径，将/ABSOLUTE/PATH/TO/GenAI_Agents替换为您存储库的绝对路径。

步骤4：重启Claude Desktop使更改生效。

在聊天框中这个锤子图标。

步骤5：尝试询问比特币价格

输入"比特币的当前价格是多少？"，将得到如下响应：

恭喜！我们已经成功应用了您的MCP服务器和工具。现在，可以尝试向 mcp_server.py 添加自己的工具。以下是一个示例：

@mcp.tool()
async def get_crypto_market_info(crypto_ids: str, currency: str = "usd") -> str:
    """
    获取一个或多个加密货币的市场信息。
    
    参数：
    - crypto_ids：逗号分隔的加密货币ID列表（例如'bitcoin,ethereum'）
    - currency：显示价值的货币（默认：'usd'）
    
    返回：
    - 包括价格、市值、交易量和价格变化的市场信息
    """
    # 构建API URL
    url = f"{COINGECKO_BASE_URL}/coins/markets"
    
    # 设置查询参数
    params = {
        "vs_currency": currency,  # 显示价值的货币
        "ids": crypto_ids,        # 逗号分隔的加密货币ID
        "order": "market_cap_desc", # 按市值排序
        "page": 1,                # 页码
        "sparkline": "false"      # 排除走势图数据
    }
    
    try:
        # 进行API调用
        async with httpx.AsyncClient() as client:
            response = await client.get(url, params=params)
            response.raise_for_status()
            
            # 解析响应
            data = response.json()
            
            # 检查是否获得任何数据
            if not data:
                return f"未找到加密货币数据：'{crypto_ids}'。请检查ID并重试。"
            
            # 格式化结果
            result = ""
            for crypto in data:
                name = crypto.get('name', 'Unknown')
                symbol = crypto.get('symbol', '???').upper()
                price = crypto.get('current_price', 'Unknown')
                market_cap = crypto.get('market_cap', 'Unknown')
                volume = crypto.get('total_volume', 'Unknown')
                price_change = crypto.get('price_change_percentage_24h', 'Unknown')
                
                result += f"{name} ({symbol}):\n"
                result += f"当前价格：{price} {currency.upper()}\n"
                result += f"市值：{market_cap} {currency.upper()}\n"
                result += f"24小时交易量：{volume} {currency.upper()}\n"
                result += f"24小时价格变化：{price_change}%\n\n"
            
            return result
            
    except Exception as e:
        return f"获取市场数据错误：{str(e)}"

使用uv run mcp_server.py重新运行您的mcp服务器，重启Claude Desktop，然后输入"What is the market data for Dogecoin and Solana?"。将得到如下响应：

通过MCP执行工具的自定义代理

构建了自己的MCP后，让我们尝试自己构建MCP Host和Client。

理解架构

在本节中，我们将构建自己的MCP Host和Client。与之前连接到Claude Desktop的方法不同，我们现在将创建自己的代理，它可以：

充当MCP Host
从我们的MCP服务器发现可用工具
基于用户查询理解何时使用哪个工具
使用适当参数执行工具
处理工具结果以提供有用响应

这种架构遵循现代AI系统中常见的模式：

发现阶段：我们的自定义主机发现哪些工具可用
规划阶段：代理根据用户查询决定使用哪个工具
执行阶段：我们的客户端连接到服务器并执行选定的工具
解释阶段：代理用自然语言解释结果

以下是简单的工作流程图：

运行代码前的重要提醒： ⚠️ 不要忘记先启动MCP服务器！⚠️ 在运行以下教程中的代理代码之前，请确保MCP服务器正在运行。否则，Agent将没有工具可发现或执行。

从设置环境和导入必要的库开始：

安装依赖

!pip install mcp anthropic

我们需要两个主要库：

MCP：处理与我们MCP服务器的客户端-服务器通信，允许我们构建主机和客户端组件
Anthropic：与Claude交互，为我们代理的推理能力提供动力

设置基础配置

# 导入必要库
import os
import json
from typing import List, Dict, Any

# 用于连接服务器的MCP库
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# Claude的Anthropic API
from anthropic import Anthropic

# 设置Anthropic API密钥
os.environ["ANTHROPIC_API_KEY"] = "your_anthropic_api_key_here"

# 初始化Anthropic客户端
client = Anthropic()

# MCP服务器路径
mcp_server_path = "absolute/path/to/your/running/mcp/server"
print("设置完成！")

我们使用MCP的stdio_client接口，它允许我们连接到作为独立进程运行并通过标准输入/输出通信的MCP服务器。这是本地开发的简单而强大的方法。通过实现MCP协议的两侧（主机和客户端），我们完全控制代理如何与MCP工具交互。

工具发现：构建我们的MCP Host

构建自定义MCP实现的第一步是创建一个能够发现MCP服务器可用工具的主机。我们的主机将充当用户、AI和可用工具之间的中介——类似于Claude Desktop的功能，但在我们的完全控制下。

让我们实现一个连接到MCP服务器并发现其工具的函数：

async def discover_tools():
    """
    连接到MCP服务器并发现可用工具。
    返回有关可用工具的信息。
    """
    # ANSI颜色代码以提高日志可见性
    BLUE = "\033[94m"
    GREEN = "\033[92m"
    RESET = "\033[0m"
    SEP = "=" * 40
    
    # 创建通过stdio连接到您的MCP服务器的服务器参数
    server_params = StdioServerParameters(
        command="python",  # 运行服务器的命令
        args=[mcp_server_path],  # MCP服务器脚本的路径
    )
    
    print(f"{BLUE}{SEP}\n🔍 发现阶段：连接到MCP服务器...{RESET}")
    
    # 通过stdio连接到服务器
    async with stdio_client(server_params) as (read, write):
        # 创建客户端会话
        async with ClientSession(read, write) as session:
            # 初始化连接
            print(f"{BLUE}📡 初始化MCP连接...{RESET}")
            await session.initialize()
            
            # 列出可用工具
            print(f"{BLUE}🔎 发现可用工具...{RESET}")
            tools = await session.list_tools()
            
            # 格式化工具信息以便查看
            tool_info = []
            for tool_type, tool_list in tools:
                if tool_type == "tools":
                    for tool in tool_list:
                        tool_info.append({
                            "name": tool.name,
                            "description": tool.description,
                            "schema": tool.inputSchema
                        })
            
            print(f"{GREEN}✅ 成功发现{len(tool_info)}个工具{RESET}")
            print(f"{SEP}")
            return tool_info

print("工具发现函数已定义")

工具执行：实现我们的MCP客户端

现在我们的主机可以发现可用工具，我们需要实现能够执行它们的客户端组件：

async def execute_tool(tool_name: str, arguments: Dict[str, Any]):
    """
    执行MCP服务器提供的特定工具。
    
    参数：
        tool_name：要执行的工具名称
        arguments：传递给工具的参数字典
        
    返回：
        执行工具的结果
    """
    # ANSI颜色代码以提高日志可见性
    BLUE = "\033[94m"
    GREEN = "\033[92m"
    YELLOW = "\033[93m"
    RESET = "\033[0m"
    SEP = "-" * 40
    
    server_params = StdioServerParameters(
        command="python",
        args=[mcp_server_path],
    )
    
    print(f"{YELLOW}{SEP}")
    print(f"⚙️ 执行阶段：运行工具'{tool_name}'")
    print(f"📋 参数：{json.dumps(arguments, indent=2)}")
    print(f"{SEP}{RESET}")
    
    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # 使用提供的参数调用特定工具
            print(f"{BLUE}📡 向MCP服务器发送请求...{RESET}")
            result = await session.call_tool(tool_name, arguments)
            
            print(f"{GREEN}✅ 工具执行完成{RESET}")
            
            # 格式化结果预览以获得更清晰的输出
            result_preview = str(result)
            if len(result_preview) > 150:
                result_preview = result_preview[:147] + "..."
                
            print(f"{BLUE}📊 结果：{result_preview}{RESET}")
            print(f"{SEP}")
            
            return result

print("工具执行函数已定义")

集成AI与我们的MCP实现

有了主机和客户端组件，我们现在需要将它们与能够对工具使用做出智能决策的AI系统集成。这是我们自定义MCP主机的"大脑"，它需要：

根据用户输入理解何时需要工具
为任务选择适当的工具
正确格式化参数
处理并解释结果

让我们实现一个协调整个过程的函数：

async def query_claude(prompt: str, tool_info: List[Dict], previous_messages=None):
    """
    向Claude发送查询并处理响应。
    
    参数：
        prompt：用户查询
        tool_info：有关可用工具的信息
        previous_messages：维护上下文的先前消息
        
    返回：
        Claude的响应，可能在执行工具后
    """
    # ANSI颜色代码以提高日志可见性
    BLUE = "\033[94m"
    GREEN = "\033[92m"
    YELLOW = "\033[93m"
    PURPLE = "\033[95m"
    RESET = "\033[0m"
    SEP = "=" * 40
    
    if previous_messages is None:
        previous_messages = []
    
    print(f"{PURPLE}{SEP}")
    print("🧠 推理阶段：使用Claude处理查询")
    print(f"🔤 查询：\"{prompt}\"")
    print(f"{SEP}{RESET}")
    
    # 为Claude格式化工具信息
    tool_descriptions = "\n\n".join([
        f"工具：{tool['name']}\n描述：{tool['description']}\n模式：{json.dumps(tool['schema'], indent=2)}"
        for tool in tool_info
    ])
    
    # 构建系统提示
    system_prompt = f"""您是一个通过MCP（模型上下文协议）访问专用工具的AI助手。
    
可用工具：
{tool_descriptions}

当您需要使用工具时，请以以下格式回复JSON对象：
{{
    "tool": "tool_name",
    "arguments": {{
        "arg1": "value1",
        "arg2": "value2"
    }}
}}

使用工具时不要包含任何其他文本，只需JSON对象。
对于常规响应，请正常回复。
"""
    
    # 从先前消息中过滤系统消息
    filtered_messages = [msg for msg in previous_messages if msg["role"] != "system"]
    
    # 构建对话消息（不包含系统消息）
    messages = filtered_messages.copy()
    
    # 添加当前用户查询
    messages.append({"role": "user", "content": prompt})
    
    print(f"{BLUE}📡 向Claude API发送请求...{RESET}")
    
    # 向Claude发送请求，系统作为顶级参数
    response = client.messages.create(
        model="claude-3-5-sonnet-20240620",
        max_tokens=4000,
        system=system_prompt,  # 系统提示作为单独参数
        messages=messages      # 仅用户和助手消息
    )
    
    # 获取Claude的响应
    claude_response = response.content[0].text
    print(f"{GREEN}✅ 收到Claude的响应{RESET}")
    
    # 尝试从响应中提取和解析JSON
    try:
        # 在响应中查找JSON模式
        import re
        json_match = re.search(r'(\{[\s\S]*\})', claude_response)
        
        if json_match:
            json_str = json_match.group(1)
            print(f"{YELLOW}🔍 在响应中检测到工具使用{RESET}")
            print(f"{BLUE}📦 提取的JSON：{json_str}{RESET}")
            
            tool_request = json.loads(json_str)
            
            if "tool" in tool_request and "arguments" in tool_request:
                tool_name = tool_request["tool"]
                arguments = tool_request["arguments"]
                
                print(f"{YELLOW}🔧 Claude想要使用工具：{tool_name}{RESET}")
                
                # 使用我们的MCP客户端执行工具
                tool_result = await execute_tool(tool_name, arguments)
                
                # 如果需要，将工具结果转换为字符串
                if not isinstance(tool_result, str):
                    tool_result = str(tool_result)
                
                # 使用工具请求和结果更新消息
                messages.append({"role": "assistant", "content": claude_response})
                messages.append({"role": "user", "content": f"工具结果：{tool_result}"})
                
                print(f"{PURPLE}🔄 获取Claude对工具结果的解释...{RESET}")
                
                # 获取Claude对工具结果的解释
                final_response = client.messages.create(
                    model="claude-3-5-sonnet-20240620",
                    max_tokens=4000,
                    system=system_prompt,
                    messages=messages
                )
                
                print(f"{GREEN}✅ 最终响应准备就绪{RESET}")
                print(f"{SEP}")
                
                return final_response.content[0].text, messages
        
    except (json.JSONDecodeError, KeyError, AttributeError) as e:
        print(f"{YELLOW}⚠️ 响应中未检测到工具使用：{str(e)}{RESET}")
    
    print(f"{GREEN}✅ 响应准备就绪{RESET}")
    print(f"{SEP}")
    
    return claude_response, messages

print("Claude查询函数已定义")

构建交互式MCP Host界面

对于完整的MCP主机实现，我们需要一个用户界面，该界面在多轮对话中维护上下文。这允许我们的主机记住先前的交互并在后续交换中基于它们构建，就像专业的MCP主机（如Claude Desktop）一样。让我们实现一个简单的聊天会话函数：

async def chat_session():
    """
    运行与AI代理的交互式聊天会话。
    """
    # ANSI颜色代码以提高日志可见性
    BLUE = "\033[94m"
    GREEN = "\033[92m"
    YELLOW = "\033[93m"
    CYAN = "\033[96m"
    BOLD = "\033[1m"
    RESET = "\033[0m"
    SEP = "=" * 50
    
    print(f"{CYAN}{BOLD}{SEP}")
    print("🤖 初始化MCP代理")
    print(f"{SEP}{RESET}")
    
    # 确保从前一个单元格定义了'tools'，或重新发现它们
    try:
        # 检查tools是否已定义且不为空
        if 'tools' not in globals() or not tools:
            print(f"{BLUE}🔍 未找到工具，正在发现可用工具...{RESET}")
            tools_local = await discover_tools()
        else:
            tools_local = tools
            
        print(f"{GREEN}✅ 代理准备就绪，有{len(tools_local)}个工具：{RESET}")
        
        # 打印可用工具以供参考
        for i, tool in enumerate(tools_local, 1):
            print(f"{YELLOW}  {i}. {tool['name']}{RESET}")
            print(f"     {tool['description'].strip()}")
        
        # 开始聊天会话
        print(f"\n{CYAN}{BOLD}{SEP}")
        print(f"💬 交互式聊天会话")
        print(f"{SEP}")
        print(f"输入'exit'或'quit'结束会话{RESET}")
        
        messages = []
        
        while True:
            # 获取用户输入
            user_input = input(f"\n{BOLD}您：{RESET} ")
            
            # 检查用户是否想退出
            if user_input.lower() in ['exit', 'quit']:
                print(f"\n{GREEN}结束聊天会话。再见！{RESET}")
                break
            
            # 使用Claude处理查询
            print(f"\n{BLUE}处理中...{RESET}")
            response, messages = await query_claude(user_input, tools_local, messages)
            
            # 显示Claude的响应
            print(f"\n{BOLD}助手：{RESET} {response}")
            
    except Exception as e:
        print(f"\n{YELLOW}⚠️ 发生错误：{str(e)}{RESET}")

print("聊天会话函数已定义。在下一个单元格中运行'await chat_session()'开始聊天。")

测试我们的MCP实现

现在让我们测试我们完整的MCP实现：

1. 首先发现可用工具

# 发现可用工具
tools = await discover_tools()
print(f"发现{len(tools)}个工具：")
for i, tool in enumerate(tools, 1):
    print(f"{i}. {tool['name']}: {tool['description']}")

2. 测试单个查询

# 使用MCP服务器的工具运行单个查询
query = "比特币的当前价格是多少？"
print(f"发送查询：{query}")

response, messages = await query_claude(query, tools)
print(f"\n助手的响应：\n{response}")

3. 启动交互式聊天

# 运行聊天会话
await chat_session()

在聊天会话中，您可以尝试询问：

“比特币的当前价格是多少？”
“狗狗币和Solana的市场数据是什么？”
“以太坊的价格怎么样？”

结论

Model Context Protocol代表了将AI模型与外部资源集成的变革性方法，解决了AI应用开发中的关键挑战：

协议优势

标准化集成：消除复杂的自定义API连接
动态工具发现：使AI能够无缝找到和使用工具
灵活通信：支持实时双向交互

技术亮点

我们的实现演示了：

构建具有专用工具的MCP服务器
创建能够动态发现和执行工具的主机
将AI模型与外部资源集成

关键参考资料

通过本教程，您已经学会了如何构建完整的MCP实现，从服务器端的工具开发到客户端的智能代理，为您的AI应用开启了无限可能。

n{response}")

### 3. 启动交互式聊天

```python
# 运行聊天会话
await chat_session()

在聊天会话中，您可以尝试询问：

“比特币的当前价格是多少？”
“狗狗币和Solana的市场数据是什么？”
“以太坊的价格怎么样？”

结论

Model Context Protocol代表了将AI模型与外部资源集成的变革性方法，解决了AI应用开发中的关键挑战：

协议优势

标准化集成：消除复杂的自定义API连接
动态工具发现：使AI能够无缝找到和使用工具
灵活通信：支持实时双向交互

技术亮点

我们的实现演示了：

构建具有专用工具的MCP服务器
创建能够动态发现和执行工具的主机
将AI模型与外部资源集成

关键参考资料

通过本教程，您已经学会了如何构建完整的MCP实现，从服务器端的工具开发到客户端的智能代理，为您的AI应用开启了无限可能。

这篇教程展示了MCP协议的强大功能和灵活性。通过掌握这些技术，您可以为您的AI代理添加各种外部能力，从数据查询到复杂的API集成，让您的应用更加智能和实用。

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2025-07-25，如有侵权请联系 cloudcommunity@tencent.com 删除

服务器

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

登录后参与评论

0 条评论

热度