MCP 模型上下文协议实战完全指南：让 AI 无缝连接外部世界

2024 年 11 月，Anthropic 推出了 MCP（Model Context Protocol），一个革命性的开放协议。它被誉为 AI 应用的「USB-C 端口」，让大模型能够标准化地连接数据库、API、文件系统等外部资源。短短几个月，OpenAI、Google DeepMind 等巨头纷纷采纳，一场 AI 集成方式的变革正在发生。

一、核心概念：什么是 MCP？

MCP（Model Context Protocol，模型上下文协议）是一个开放标准和开源框架，旨在标准化人工智能系统与外部工具、数据源的集成方式。在 MCP 出现之前，开发者需要为每个数据源构建定制连接器，这导致了复杂的「N×M」数据集成问题。MCP 的出现彻底改变了这一局面。

1.1 为什么需要 MCP？

传统 AI 应用开发面临三大痛点：

• 集成复杂：每个数据源都需要单独开发连接器，维护成本极高
• 知识滞后：大模型只能访问训练数据，无法获取实时信息
• 能力受限：AI 只能「对话」，无法执行实际操作

MCP 通过标准化协议解决了这些问题。就像 USB-C 接口统一了硬件连接，MCP 统一了 AI 与外部世界的交互方式。

1.2 MCP 的核心架构

MCP 采用客户端-服务器架构，主要包含以下组件：

• MCP Host（主机）：包含大模型的 AI 应用，如 Claude Desktop、Cursor 等
• MCP Client（客户端）：负责主机与服务器之间的通信
• MCP Server（服务器）：提供具体功能的外部服务，如数据库、API
• Transport Layer（传输层）：使用 JSON-RPC 2.0 进行通信

工作流程如下：当用户向 AI 提问时，AI 判断是否需要外部工具，通过 MCP Client 调用对应的 MCP Server，Server 执行操作后返回结果，AI 将信息整合到回答中。

二、核心功能：MCP 能做什么？

2.1 三大核心能力

MCP Server 可以提供三种主要功能：

1. Resources（资源）

只读的文件类数据，如 API 响应、文件内容、数据库记录等。AI 可以读取这些资源获取上下文信息。

2. Tools（工具）

可执行的函数，AI 可以调用这些工具执行操作，如发送邮件、创建文件、查询天气等。需要用户批准。

3. Prompts（提示）

预编写的提示模板，帮助用户快速完成特定任务，如代码审查、数据分析等。

2.2 实际应用场景

MCP 的应用场景非常广泛：

• 智能助手：连接 Google 日历、Notion，成为个性化助理
• 企业应用：连接多个数据库，进行数据分析与报告生成
• 开发工具：连接代码仓库、CI/CD 系统，实现自动化开发
• 创意应用：在 Blender 中创建 3D 设计并打印

三、实战案例：从零构建 MCP Server

下面我们通过一个完整的实战案例，演示如何构建一个天气查询 MCP Server。

3.1 环境准备

首先安装必要的依赖：

# 安装 MCP SDK
pip install mcp

# 或者使用 Python 官方 SDK
pip install mcp[cli]

3.2 构建天气 MCP Server

创建一个简单的天气服务器，提供两个工具：获取天气预警和天气预报。

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
import httpx
import asyncio

# 创建 MCP Server 实例
server = Server("weather-server")

# 国家天气服务 API 基础 URL
NWS_API_BASE = "https://api.weather.gov"

@server.list_tools()
async def list_tools():
    """定义服务器提供的工具列表"""
    return [
        Tool(
            name="get_alerts",
            description="获取美国各州的天气预警信息",
            inputSchema={
                "type": "object",
                "properties": {
                    "state": {
                        "type": "string",
                        "description": "美国州代码，如 CA、NY"
                    }
                },
                "required": ["state"]
            }
        ),
        Tool(
            name="get_forecast",
            description="获取指定位置的天气预报",
            inputSchema={
                "type": "object",
                "properties": {
                    "latitude": {
                        "type": "number",
                        "description": "纬度"
                    },
                    "longitude": {
                        "type": "number",
                        "description": "经度"
                    }
                },
                "required": ["latitude", "longitude"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    """执行工具调用"""
    async with httpx.AsyncClient() as client:
        if name == "get_alerts":
            state = arguments["state"]
            url = f"{NWS_API_BASE}/alerts?area={state}"
            response = await client.get(url)
            data = response.json()
            
            # 格式化预警信息
            alerts = []
            for alert in data.get("features", []):
                alerts.append(f"{alert['properties']['event']}: {alert['properties']['headline']}")
            
            return [TextContent(type="text", text="\n".join(alerts) if alerts else "无预警信息")]
            
        elif name == "get_forecast":
            lat = arguments["latitude"]
            lon = arguments["longitude"]
            
            # 获取预报点 URL
            points_url = f"{NWS_API_BASE}/points/{lat},{lon}"
            points_response = await client.get(points_url)
            points_data = points_response.json()
            
            # 获取详细预报
            forecast_url = points_data["properties"]["forecast"]
            forecast_response = await client.get(forecast_url)
            forecast_data = forecast_response.json()
            
            # 提取预报信息
            periods = forecast_data["properties"]["periods"][:3]
            forecasts = []
            for period in periods:
                forecasts.append(f"{period['name']}: {period['temperature']}°{period['temperatureUnit']}, {period['shortForecast']}")
            
            return [TextContent(type="text", text="\n".join(forecasts))]

async def main():
    """启动 MCP 服务器"""
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream)

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

3.3 配置 Claude Desktop 连接 MCP Server

在 Claude Desktop 中配置 MCP Server：

{
  "mcpServers": {
    "weather": {
      "command": "python",
      "args": ["/path/to/weather_server.py"]
    }
  }
}

配置文件位置：

• macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
• Windows: %APPDATA%\Claude\claude_desktop_config.json

3.4 测试使用

重启 Claude Desktop 后，你可以这样使用：

用户：加州有什么天气预警吗？

Claude：[调用 get_alerts 工具]
当前加州有以下天气预警：
- Flood Warning: 塞拉山脉地区洪水预警
- Wind Advisory: 强风预警

四、进阶应用：构建企业级 MCP Server

4.1 连接数据库

构建一个可以查询企业数据库的 MCP Server：

from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent, Resource
import sqlite3

server = Server("database-server")

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="query_sales",
            description="查询销售数据",
            inputSchema={
                "type": "object",
                "properties": {
                    "product": {"type": "string"},
                    "date_from": {"type": "string"},
                    "date_to": {"type": "string"}
                },
                "required": ["date_from", "date_to"]
            }
        )
    ]

@server.list_resources()
async def list_resources():
    """列出可用的数据库表"""
    return [
        Resource(
            uri="db://schema",
            name="数据库结构",
            mimeType="text/plain"
        )
    ]

@server.read_resource()
async def read_resource(uri: str):
    if uri == "db://schema":
        # 返回数据库结构信息
        schema_info = """表结构：
- sales: 销售记录表
  - id, product, amount, date, region
- products: 产品信息表
  - id, name, category, price"""
        return schema_info

@server.call_tool()
async def call_tool(name: str, arguments: dict):
    conn = sqlite3.connect("company.db")
    cursor = conn.cursor()
    
    if name == "query_sales":
        query = """
        SELECT product, SUM(amount) as total
        FROM sales
        WHERE date BETWEEN ? AND ?
        GROUP BY product
        """
        cursor.execute(query, (arguments["date_from"], arguments["date_to"]))
        results = cursor.fetchall()
        
        report = "销售汇总报告：\n"
        for row in results:
            report += f"- {row[0]}: {row[1]:,.2f} 元\n"
        
        return [TextContent(type="text", text=report)]
    
    conn.close()

async def main():
    async with stdio_server() as streams:
        await server.run(*streams)

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

4.2 安全与权限控制

企业级应用需要考虑安全性：

from functools import wraps

def require_permission(permission: str):
    """权限检查装饰器"""
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # 从上下文获取用户信息
            context = server.context
            user_permissions = context.get("permissions", [])
            
            if permission not in user_permissions:
                return [TextContent(
                    type="text", 
                    text="错误：您没有执行此操作的权限"
                )]
            
            return await func(*args, **kwargs)
        return wrapper
    return decorator

@server.call_tool()
@require_permission("sales:read")
async def query_sensitive_data(arguments: dict):
    # 只有拥有 sales:read 权限的用户才能执行
    pass

五、MCP 生态系统

5.1 官方预构建 Server

Anthropic 和社区已经提供了大量预构建的 MCP Server：

• Filesystem Server：安全地访问本地文件系统
• PostgreSQL Server：连接 PostgreSQL 数据库
• GitHub Server：GitHub API 集成
• Google Drive Server：访问 Google Drive 文件
• Slack Server：发送和读取 Slack 消息

5.2 客户端支持

目前支持 MCP 的客户端包括：

• Claude Desktop：Anthropic 官方客户端
• Cursor：AI 编程助手
• Zed：高性能编辑器
• 源代码集成：可在 Python、TypeScript 应用中直接使用

六、最佳实践

6.1 工具设计原则

• 单一职责：每个工具只做一件事
• 清晰描述：工具描述要详细，帮助 AI 理解何时调用
• 参数验证：使用 JSON Schema 严格验证输入
• 错误处理：提供友好的错误信息，便于调试

6.2 性能优化

• 缓存策略：对频繁访问的数据进行缓存
• 异步操作：使用异步 I/O 提高并发性能
• 连接池：复用数据库和 API 连接

总结

MCP 作为 AI 集成的标准化协议，正在改变 AI 应用的开发方式。它让 AI 从「会说话的百科全书」进化为「能干活的智能助手」。无论是个人开发者还是企业团队，都可以通过 MCP 快速构建强大的 AI 应用。

关键要点：

• MCP 统一了 AI 与外部工具的集成方式，降低了开发复杂度
• 三大核心能力：Resources（资源）、Tools（工具）、Prompts（提示）
• 支持 Claude Desktop、Cursor 等主流客户端
• 丰富的预构建 Server 生态，开箱即用

随着 OpenAI、Google 等巨头的加入，MCP 很可能成为 AI 集成的行业标准。现在开始学习 MCP，就是为未来的 AI 应用开发做好准备。

本文链接：https://www.kkkliao.cn/?id=949 转载需授权！