MCP协议：AI智能体生态的新基础设施

基于腾讯云 AI技术落地实战经验 征文

一、引言：AI智能体时代的数据孤岛困境

2025年被业界称为"智能体元年"。随着大模型能力的飞速提升，AI Agent（智能体）已成为AI应用的主流形态。然而，一个根本性的问题始终困扰着开发者：如何让AI智能体高效地与外部世界交互？

传统方案是"一对一定制"：每个AI应用都需要为每个数据源或工具单独开发集成代码。AI编程助手需要集成GitHub、GitLab、Jira；数据分析Agent需要连接数据库、BI工具、云存储……这种碎片化的集成方式造成了严重的重复劳动和数据孤岛问题。

根据Anthropic的统计数据，一个典型的企业级AI应用往往需要对接 10-50个 不同的数据源和工具，开发周期长达 数周甚至数月。

二、MCP协议的诞生与核心理念

Model Context Protocol（MCP） 是由Anthropic于2024年底推出的开放协议，旨在为AI模型与外部数据源、工具的交互提供统一标准。2025年6月18日，MCP发布了涵盖安全性增强、功能补完与规范简化的重大更新，引入了结构化数据验证、elicitation机制和更严格的版本协商策略。

MCP的核心设计理念是 “一次开发，处处连接” —— 开发者只需在Agent端实现一次MCP客户端，即可无缝对接任何支持MCP的服务端。这种"即插即用"的模式彻底改变了AI应用的集成范式。

2.1 MCP协议的核心架构

MCP采用JSON-RPC 2.0消息格式，建立了三方通信模型：

角色 描述 职责
Host LLM应用程序 发起连接，协调多个Client
Client Host内部的连接器 管理与服务器的连接和通信
Server 外部数据源/工具服务 提供上下文和能力

2.2 协议基础信息

  • 传输方式：支持 stdio（标准输入输出）和 SSE（Server-Sent Events）
  • 消息格式：JSON-RPC 2.0
  • 状态管理：有状态连接，支持长会话
  • 能力协商：客户端与服务端动态协商支持的功能

三、MCP协议的核心组件

MCP协议定义了三大核心能力，这些能力共同构成了AI智能体与外部世界交互的完整方案。

3.1 资源（Resources）

资源是AI可以读取的外部数据，如文件、数据库、API响应等。每个资源都有唯一的URI标识。

服务端示例（Python + FastMCP）：

from mcp.server import FastMCP
from mcp.types import Resource, TextResourceContents

# 创建MCP服务器
mcp = FastMCP("my-file-server")

@mcp.list_resources()
async def list_resources() -> list[Resource]:
    """列出所有可用资源"""
    return [
        Resource(
            uri="file:///workspace/README.md",
            name="readme",
            description="项目README文件",
            mimeType="text/markdown"
        ),
        Resource(
            uri="file:///workspace/config.json",
            name="config",
            description="应用配置文件",
            mimeType="application/json"
        ),
        Resource(
            uri="file:///workspace/src/",
            name="source-code",
            description="源代码目录",
            mimeType="text/plain"
        )
    ]

@mcp.read_resource()
async def read_resource(uri: str) -> str | bytes:
    """读取指定资源"""
    import os
    path = uri.replace("file://", "")
    
    if os.path.isdir(path):
        # 返回目录列表
        files = os.listdir(path)
        return "\n".join(files)
    
    with open(path, "r", encoding="utf-8") as f:
        return f.read()

if __name__ == "__main__":
    mcp.run(transport="stdio")

3.2 工具（Tools）

工具是MCP最重要的能力，允许AI调用外部函数执行操作。

服务端示例（TypeScript）：

import { Server } from "@modelcontextprotocol/sdk/server/index.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { CallToolRequestSchema, ListToolsRequestSchema } from "@modelcontextprotocol/sdk/types.js";

const server = new Server({
    name: "github-mcp-server",
    version: "1.0.0"
}, { capabilities: { tools: {} }});

server.setRequestHandler(ListToolsRequestSchema, async () => {
    return {
        tools: [
            {
                name: "github_get_issues",
                description: "获取GitHub仓库的Issue列表",
                inputSchema: {
                    type: "object",
                    properties: {
                        owner: { type: "string", description: "仓库所有者" },
                        repo: { type: "string", description: "仓库名称" },
                        state: { type: "string", enum: ["open", "closed", "all"], default: "open" }
                    },
                    required: ["owner", "repo"]
                }
            }
        ]
    };
});

server.setRequestHandler(CallToolRequestSchema, async (request) => {
    const { name, arguments: args } = request.params;
    
    if (name === "github_get_issues") {
        const response = await fetch(
            `https://api.github.com/repos/${args.owner}/${args.repo}/issues`,
            { headers: { "Authorization": `Bearer ${process.env.GITHUB_TOKEN}` }}
        );
        const issues = await response.json();
        return {
            content: [{ type: "text", text: JSON.stringify(issues, null, 2) }]
        };
    }
});

客户端调用示例（Python）：

import asyncio
from mcp import ClientSession, StdioServerParameters

async def use_github_tools():
    server_params = StdioServerParameters(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-github"],
        env={"GITHUB_TOKEN": "your-github-token"}
    )
    
    async with ClientSession(server_params) as session:
        await session.initialize()
        
        # 获取可用工具列表
        tools = await session.list_tools()
        print(f"可用工具：{len(tools.tools)}个")
        
        # 调用工具获取Issue列表
        result = await session.call_tool("github_get_issues", {
            "owner": "anthropic",
            "repo": "mcp",
            "state": "open"
        })
        
        for content in result.content:
            if content.type == "text":
                issues = json.loads(content.text)
                for issue in issues[:5]:
                    print(f"#{issue['number']}: {issue['title']}")

asyncio.run(use_github_tools())

3.3 提示模板（Prompts）

提示模板允许服务端提供可复用的提示词片段。

from mcp.types import Prompt, PromptArgument, PromptMessage, TextContent, GetPromptResult

@mcp.list_prompts()
async def list_prompts() -> list[Prompt]:
    return [
        Prompt(
            name="review_code",
            description="代码审查模板",
            arguments=[
                PromptArgument(name="repository", description="代码仓库路径", required=True),
                PromptArgument(name="language", description="编程语言", required=False)
            ]
        )
    ]

@mcp.get_prompt()
async def get_prompt(name: str, arguments: dict) -> GetPromptResult:
    if name == "review_code":
        return GetPromptResult(
            description="代码审查模板",
            messages=[
                PromptMessage(
                    role="user",
                    content=TextContent(
                        type="text",
                        text=f"""请审查代码仓库 {arguments['repository']}：

### 审查检查清单
- [ ] 代码风格一致性
- [ ] 安全性检查
- [ ] 性能优化建议
- [ ] 测试覆盖率

请给出详细审查报告。"""
                    )
                )
            ]
        )

四、代码执行：MCP的效率优化利器

根据Anthropic在2025年发布的技术博客，随着MCP使用规模扩大，存在两个典型性能问题：

4.1 问题一：工具定义占用大量上下文

当Agent连接数百甚至数千个工具时，工具定义的描述会占用数十万Tokens。

4.2 问题二：中间结果重复传递

传统模式下，每个工具调用的结果都需要经过模型处理。例如：将Google Drive的会议记录写入Salesforce，完整记录要在上下文窗口中传递两次。

传统方式的问题：

Context Window:
├── System Prompt (500 tokens)
├── Tool Definitions (2000 tools × 100 tokens = 200,000 tokens)
├── User Query (100 tokens)
└── Response (2000 tokens)

4.3 解决方案：代码执行模式

将MCP服务器呈现为代码API，Agent直接编写代码交互：

# 优化后的代码执行方式
# Token消耗：仅需描述代码功能（约200 tokens）

result = await session.execute_code("""
import requests

# 获取Google Drive文档
G_DRIVE_TOKEN = os.getenv('GDRIVE_TOKEN')
SF_TOKEN = os.getenv('SALESFORCE_TOKEN')

doc_response = requests.get(
    f"https://www.googleapis.com/drive/v3/files/{doc_id}?alt=media",
    headers={"Authorization": f"Bearer {G_DRIVE_TOKEN}"}
)
doc_content = doc_response.text

# 直接写入Salesforce（数据不经过模型上下文）
requests.patch(
    f"https://api.salesforce.com/services/data/v59.0/sobjects/Lead/{lead_id}",
    json={"Notes": doc_content},
    headers={"Authorization": f"Bearer {SF_TOKEN}"}
)

return "Document transferred"
""")

效果对比：

方案 Token消耗 延迟 说明
传统工具调用 ~10,000+ 高 数据经过模型多次传递
代码执行模式 ~500 低 仅在模型中描述操作意图

4.4 实现代码执行MCP服务器

from mcp.server import FastMCP
import subprocess
import tempfile
import os

mcp = FastMCP("code-executor")

@mcp.tool()
async def execute_python(code: str, timeout: int = 30) -> str:
    """在安全沙箱中执行Python代码"""
    with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
        f.write(code)
        temp_file = f.name
    
    try:
        result = subprocess.run(
            ['docker', 'run', '--rm', '-v', f'{temp_file}:/script.py',
             'python:3.11-slim', 'python', '/script.py'],
            capture_output=True, text=True, timeout=timeout
        )
        return result.stdout if result.returncode == 0 else result.stderr
    finally:
        os.unlink(temp_file)

@mcp.tool()
async def execute_sql(connection_string: str, query: str) -> str:
    """执行SQL查询（仅SELECT）"""
    if not query.strip().lower().startswith('select'):
        return "错误：仅支持SELECT查询"
    
    import sqlalchemy
    try:
        engine = sqlalchemy.create_engine(connection_string)
        with engine.connect() as conn:
            result = conn.execute(sqlalchemy.text(query))
            return str(result.fetchall())
    except Exception as e:
        return f"查询错误: {str(e)}"

五、实战案例：AI自动化代码审查系统

5.1 系统架构

┌─────────────────────────────────────────────────────────┐
│                    MCPReviewAgent                        │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐   │
│  │ GitHub MCP   │  │ Filesystem   │  │ Claude API   │   │
│  │   Server     │  │   MCP Server │  │   Client     │   │
│  └──────┬───────┘  └──────┬───────┘  └──────┬───────┘   │
└─────────┼────────────────┼────────────────┼───────────┘
          │                │                │
          ▼                ▼                ▼
    ┌──────────┐    ┌──────────┐    ┌──────────┐
    │  GitHub  │    │ 本地文件 │    │ Anthropic│
    │   API    │    │   系统   │    │   API    │
    └──────────┘    └──────────┘    └──────────┘

5.2 核心代码实现

# mcp_review_agent.py
import asyncio
import os
import json
from typing import Optional, List
from dataclasses import dataclass
from datetime import datetime

from mcp import ClientSession, StdioServerParameters
from anthropic import Anthropic


@dataclass
class ReviewResult:
    pr_number: int
    pr_title: str
    overall_score: int
    issues_found: List[dict]
    suggestions: List[str]
    review_comment: str
    reviewed_at: datetime


class MCPReviewAgent:
    def __init__(self, github_token: str, anthropic_api_key: str):
        self.github_token = github_token
        self.claude = Anthropic(api_key=anthropic_api_key)
        self.github_params = StdioServerParameters(
            command="npx",
            args=["-y", "@modelcontextprotocol/server-github"],
            env={"GITHUB_TOKEN": github_token}
        )
    
    async def review_pull_request(self, owner: str, repo: str, pr_number: int) -> ReviewResult:
        async with ClientSession(self.github_params) as session:
            await session.initialize()
            
            # 1. 获取PR详情
            pr_info = await self._get_pr_details(session, owner, repo, pr_number)
            
            # 2. 获取代码变更
            changed_files = await self._get_changed_files(session, owner, repo, pr_number)
            file_diffs = await self._get_file_diffs(changed_files)
            
            # 3. AI分析
            analysis = await self._analyze_with_claude(pr_info, file_diffs)
            
            # 4. 发布审查评论
            review_comment = self._generate_review_comment(analysis)
            await self._post_review_comment(session, owner, repo, pr_number, review_comment)
            
            return ReviewResult(
                pr_number=pr_number,
                pr_title=pr_info.get("title", ""),
                overall_score=analysis.get("overall_score", 70),
                issues_found=analysis.get("issues", []),
                suggestions=analysis.get("suggestions", []),
                review_comment=review_comment,
                reviewed_at=datetime.now()
            )
    
    async def _analyze_with_claude(self, pr_info: dict, file_diffs: List[dict]) -> dict:
        prompt = f"""作为代码审查专家，请审查以下PR：

标题：{pr_info.get('title')}
作者：{pr_info.get('user', {}).get('login')}

代码变更：
```json
{json.dumps(file_diffs, indent=2)}

请从以下维度审查：

  1. 代码质量：命名、风格、复杂度
  2. 安全性：注入、认证、敏感数据
  3. 性能：算法、资源、优化
  4. 测试：覆盖、边界、错误处理

返回JSON格式：{{“overall_score”: 0-100, “issues”: […], “suggestions”: […]}}
“”"

    response = self.claude.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=4000,
        messages=[{"role": "user", "content": prompt}]
    )
    
    content = response.content[0].text
    json_start = content.find('{')
    json_end = content.rfind('}') + 1
    return json.loads(content[json_start:json_end])

def _generate_review_comment(self, analysis: dict) -> str:
    score = analysis.get("overall_score", 70)
    emoji = "🟢" if score >= 80 else "🟡" if score >= 60 else "🔴"
    
    return f"""## {emoji} AI代码审查报告

综合评分：{score}/100

{analysis.get(“summary”, “”)}

问题 ({len(analysis.get(“issues”, []))}个)

“”" + “\n”.join([
f"- {i[‘file’]}:{i.get(‘line’, ‘N/A’)} - {i[‘description’]}"
for i in analysis.get(“issues”, [])
])

使用示例

async def main():
agent = MCPReviewAgent(
github_token=os.getenv(“GITHUB_TOKEN”),
anthropic_api_key=os.getenv(“ANTHROPIC_API_KEY”)
)

result = await agent.review_pull_request("myorg", "myrepo", 42)
print(f"审查完成！评分：{result.overall_score}/100")
print(f"发现问题：{len(result.issues_found)}个")

asyncio.run(main())


---

## 六、2025年6月MCP协议重大更新

### 6.1 结构化数据验证

```python
# 新增：输入输出验证
from mcp.types import Tool, StringSchema, NumberSchema

validated_tool = Tool(
    name="query_database",
    description="查询数据库",
    inputSchema={
        "type": "object",
        "properties": {
            "sql": {
                "type": "string",
                "pattern": "^SELECT.*FROM.*$",  # 限制只能执行查询
                "maxLength": 1000
            }
        },
        "required": ["sql"]
    }
)

6.2 Elicitation机制

允许AI在关键决策时请求人类确认：

{
  "method": "tools/list",
  "params": {
    "elicitation": {
      "message": "此操作将删除生产环境数据库，确定要继续吗？",
      "acceptOptions": ["确认删除", "取消操作"],
      "context": {
        "resource": "production-db",
        "action": "delete"
      }
    }
  }
}

6.3 版本协商策略

# 客户端版本协商
client_capabilities = {
    "protocolVersion": "2025-06-18",
    "supports": ["sampling", "roots", "elicitation"],
    "features": {
        "tools": {"dynamicRegistration": True},
        "resources": {"subscription": True}
    }
}

七、未来展望

MCP协议的诞生标志着AI应用开发进入了一个新阶段。随着2025年6月重大更新的发布，协议在安全性、可靠性和易用性上都有了质的飞跃。

可以预见，MCP将成为AI智能体时代的 “USB-C接口” —— 一个统一、开放、通用的连接标准，让AI与世界的交互变得像插拔设备一样简单。

对于开发者而言：

  • 降低集成成本：一次开发，对接无限数据源
  • 提升开发效率：社区已有数千个开源MCP服务器可供复用
  • 构建生态壁垒：支持MCP将成为AI产品的标配能力

参考资料

  1. Anthropic官方工程博客：Code execution with MCP
  2. MCP协议规范：modelcontextprotocol.io/specification
  3. 腾讯云开发者社区：重塑AI应用新范式：Anthropic MCP架构深度解析
  4. MCP官方SDK：GitHub - modelcontextprotocol
# 人工智能
Logo
2048 AI社区

有“AI”的1024 = 2048，欢迎大家加入2048 AI社区

更多推荐

cover

大模型“赛博算命”大考：DeepSeek、GPT-5、Gemini 谁才是真正的玄学宗师？（附 BaziQA 完整榜单）

avatar 2048 AI社区
cover

Spring AI 1.x 系列【27】Chat Memory API：让 LLM 拥有上下文记忆能力

avatar 2048 AI社区
cover

花9块9每月省22小时300块纪要费？2026年实时录音转文字怎么弄高性价比方案不看血亏

avatar 2048 AI社区