第15节：什么是MCP【模型上下文协议:连接AI与工具的开放桥梁】
目录
一、前言
在人工智能迅猛发展的今天，大型语言模型（LLM）已经展现出令人惊叹的对话、推理和生成能力。然而，当我们尝试将这些模型集成到实际应用中时，往往面临一个核心挑战：如何让模型有效地与外部系统、工具和数据源进行交互？传统的方法如Function Calling虽然提供了解决方案，但存在诸多限制：工具定义格式不统一、模型适配复杂、扩展性差等。
2023年底，Anthropic公司提出了模型上下文协议（Model Context Protocol，简称MCP），旨在为LLM与外部工具之间的通信建立一个开放、标准化的接口规范。MCP不仅仅是一个技术规范，更是一种设计哲学——它试图解决AI生态系统中的互操作性问题，让开发者能够以统一的方式为不同模型提供工具和上下文信息。
本文将深入浅出地解析MCP的核心原理、技术架构、应用场景和实践方法。无论你是AI应用开发者、企业技术决策者，还是对AI技术感兴趣的学习者，理解MCP都将帮助你更好地把握AI工具化、实用化的发展方向。我们将从基础概念入手，通过实际代码示例展示MCP的实现方式，并探讨其在行业中的价值与应用前景。
二、什么是MCP
2.1 MCP原理
模型上下文协议（Model Context Protocol，MCP） 是一种开放协议，定义了大语言模型（LLM）如何与外部工具、数据源和系统进行标准化的交互。MCP的核心思想是解耦和标准化：将工具的定义、发现、调用和结果返回等过程规范化，使得不同的模型、不同的工具能够在统一的框架下协同工作。
1. 基本工作原理
MCP的工作原理可以概括为以下几个关键步骤：
┌─────────────┐ 请求工具列表 ┌─────────────┐
│ │───────────────────▶│ │
│ 模型 │ │ MCP服务器 │
│ (客户端) │◀───────────────────│ (工具提供方) │
│ │ 返回工具定义 │ │
└──────┬──────┘ └──────┬──────┘
│ │
│ 调用工具(参数) │
│─────────────────────────────────▶│
│ │
│ 返回结果/错误 │
│◀─────────────────────────────────│
│ │2. 核心设计原则
MCP基于以下几个核心设计原则：
-
协议无关性：不绑定特定的传输协议（支持HTTP、WebSocket、stdio等）
-
双向通信：支持服务器主动推送和客户端主动请求
-
强类型定义：使用JSON Schema等标准定义工具接口
-
可扩展性：易于添加新类型的工具和功能
-
安全性：支持认证、授权和访问控制
3. 通信模式
MCP支持三种主要的通信模式：
# 1. 请求-响应模式（同步）
模型 → 请求工具 → 服务器 → 执行工具 → 返回结果 → 模型
# 2. 服务器推送模式（异步）
服务器 → 推送可用工具列表 → 模型
服务器 → 推送上下文更新 → 模型
# 3. 双向流模式（实时交互）
模型 ⇄ 持续交互 ⇄ 服务器4. 状态管理
MCP采用无状态设计，每次请求都是独立的。但通过上下文标识符（context_id）支持有状态的会话：
# 状态管理示例
{
"type": "tool_call",
"tool": "database_query",
"parameters": {"query": "SELECT * FROM users"},
"context_id": "session_123456", # 会话标识
"metadata": {
"user_id": "user_789",
"timestamp": "2024-01-15T10:30:00Z"
}
}2.2 MCP组成元素
MCP协议由几个核心组件构成，每个组件都有明确的职责和接口定义。
1. 工具（Tools）
工具是MCP的核心概念，代表模型可以调用的外部功能。每个工具都有明确的定义：
# 工具定义示例
{
"name": "weather_query", # 工具名称
"description": "查询指定城市的天气情况", # 工具描述
"input_schema": { # 输入参数模式
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，如：北京、上海"
},
"date": {
"type": "string",
"description": "日期，格式：YYYY-MM-DD，默认为今天"
}
},
"required": ["city"]
},
"output_schema": { # 输出结果模式
"type": "object",
"properties": {
"temperature": {"type": "number"},
"condition": {"type": "string"},
"humidity": {"type": "number"},
"wind_speed": {"type": "number"}
}
}
}2. 资源（Resources）
资源是模型可以访问的只读数据源，如文件、数据库视图、API端点等：
# 资源定义示例
{
"uri": "file:///data/reports/q4_2023.pdf", # 资源标识符
"mimeType": "application/pdf", # 媒体类型
"name": "Q4 2023财务报告", # 资源名称
"description": "2023年第四季度财务报告PDF文档" # 资源描述
}3. 提示模板（Prompt Templates）
提示模板是预定义的提示词结构，帮助模型更好地使用工具：
# 提示模板示例
{
"name": "data_analysis_prompt",
"template": """请分析以下数据：
数据集：{{dataset_name}}
时间范围：{{start_date}} 到 {{end_date}}
分析维度：{{dimensions}}
请使用数据分析工具处理，并给出洞察。""",
"variables": ["dataset_name", "start_date", "end_date", "dimensions"]
}4. 服务器（Server）
MCP服务器是工具的提供者和管理者，负责：
-
注册和管理工具
-
处理工具调用请求
-
管理资源访问
-
维护会话状态
# 简化的MCP服务器类结构
class MCPServer:
"""MCP服务器基类"""
def __init__(self):
self.tools = {} # 工具注册表
self.resources = {} # 资源注册表
self.sessions = {} # 会话管理
def register_tool(self, tool_name, tool_function, schema):
"""注册工具"""
self.tools[tool_name] = {
"function": tool_function,
"schema": schema
}
def handle_request(self, request):
"""处理请求"""
if request["type"] == "list_tools":
return self.list_tools()
elif request["type"] == "call_tool":
return self.call_tool(
request["tool"],
request.get("parameters", {})
)
# ... 其他请求类型
def list_tools(self):
"""列出所有可用工具"""
return [tool["schema"] for tool in self.tools.values()]
def call_tool(self, tool_name, parameters):
"""调用工具"""
if tool_name not in self.tools:
raise ValueError(f"工具未找到: {tool_name}")
tool = self.tools[tool_name]
return tool["function"](**parameters)5. 客户端（Client）
MCP客户端是模型或应用程序，负责：
-
发现可用工具
-
调用工具功能
-
处理返回结果
-
管理上下文
# 简化的MCP客户端类结构
class MCPClient:
"""MCP客户端"""
def __init__(self, server_url):
self.server_url = server_url
self.available_tools = []
self.context = {}
def discover_tools(self):
"""发现服务器上的可用工具"""
# 发送请求获取工具列表
response = self._send_request({
"type": "list_tools"
})
self.available_tools = response.get("tools", [])
return self.available_tools
def call_tool(self, tool_name, parameters):
"""调用工具"""
return self._send_request({
"type": "call_tool",
"tool": tool_name,
"parameters": parameters
})
def _send_request(self, request):
"""发送请求到服务器"""
# 实际实现会使用HTTP、WebSocket等
# 这里简化为直接调用
pass6. 传输层（Transport）
MCP支持多种传输协议，使得它可以在不同环境中使用：
-
stdio：标准输入输出，用于命令行工具
-
HTTP：基于RESTful API
-
WebSocket：全双工通信，适合实时应用
-
gRPC：高性能RPC框架
# 传输层适配器示例
class TransportAdapter:
"""传输层适配器基类"""
def send(self, message):
"""发送消息"""
pass
def receive(self):
"""接收消息"""
pass
def close(self):
"""关闭连接"""
pass
class HTTPTransport(TransportAdapter):
"""HTTP传输适配器"""
def __init__(self, base_url):
self.base_url = base_url
self.session = requests.Session()
def send(self, message):
"""通过HTTP发送消息"""
response = self.session.post(
f"{self.base_url}/mcp",
json=message
)
return response.json()2.3 MCP和Function Call对比
Function Calling是OpenAI在2023年6月推出的功能，允许GPT模型调用开发者定义的函数。虽然MCP和Function Calling都旨在连接模型与外部工具，但它们在设计理念、技术实现和应用场景上存在显著差异。
1. 设计理念对比
|
特性 |
Function Calling |
MCP |
|---|---|---|
|
设计目标 |
为OpenAI API设计的专有功能 |
开放、跨模型的通用协议 |
|
标准化程度 |
厂商特定实现 |
开放标准，多厂商支持 |
|
扩展性 |
有限，绑定OpenAI生态系统 |
高，支持多种扩展 |
|
社区驱动 |
由OpenAI控制 |
由社区推动和发展 |
2. 技术架构对比
# Function Calling示例（OpenAI专用）
def get_current_weather(location, unit="celsius"):
"""获取当前天气"""
return {"temperature": 22, "unit": unit, "condition": "晴朗"}
functions = [
{
"name": "get_current_weather",
"description": "获取当前天气",
"parameters": {
"type": "object",
"properties": {
"location": {"type": "string"},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
},
"required": ["location"]
}
}
]
# 调用方式特定于OpenAI API
response = openai.ChatCompletion.create(
model="gpt-3.5-turbo",
messages=[{"role": "user", "content": "北京天气怎么样？"}],
functions=functions,
function_call="auto"
)# MCP示例（模型无关）
# 工具定义（与Function Calling类似，但更标准化）
weather_tool = {
"name": "weather_query",
"description": "查询天气",
"input_schema": {
"type": "object",
"properties": {
"location": {"type": "string"},
"unit": {"type": "string", "enum": ["celsius", "fahrenheit"]}
},
"required": ["location"]
}
}
# 通过标准MCP协议调用，不依赖特定模型API
# 可以在Claude、GPT、本地模型等多种模型中使用3. 工具发现机制对比
-
Function Calling：工具定义在每次请求中发送，模型需要重新解析
-
MCP：工具在连接时注册，模型可以缓存和复用工具定义
# Function Calling：每次请求都包含工具定义
request = {
"model": "gpt-4",
"messages": [...],
"functions": [large_tool_definitions], # 每次都要发送
"function_call": "auto"
}
# MCP：工具在会话开始时注册
# 连接建立时
client.discover_tools() # 获取工具列表，缓存起来
# 后续请求
client.call_tool("weather_query", {"location": "北京"})
# 不需要重复发送工具定义4. 实时性对比
-
Function Calling：主要支持请求-响应模式
-
MCP：支持服务器主动推送，实时性更好
# MCP支持服务器主动推送新工具
def on_tool_registered(server, tool_definition):
"""当新工具注册时，主动推送给所有连接的客户端"""
for client in server.connected_clients:
client.send({
"type": "tool_added",
"tool": tool_definition
})
# 客户端可以实时接收新工具，无需重新连接5. 适用场景对比
|
场景 |
推荐使用 |
|---|---|
|
仅使用OpenAI模型 |
Function Calling |
|
多模型支持需求 |
MCP |
|
工具需要动态注册/注销 |
MCP |
|
简单的一次性工具调用 |
Function Calling |
|
复杂的企业级集成 |
MCP |
|
需要工具组合和流程编排 |
MCP |
6. 性能考虑
-
网络开销：MCP减少了重复传输工具定义的开销
-
连接管理：MCP支持长连接，减少握手开销
-
缓存机制：MCP客户端可以缓存工具定义，提高响应速度
2.4 MCP应用场景
MCP的标准化和开放性使其在多种应用场景中具有显著优势。以下是MCP的主要应用领域：
1. 企业知识库集成
企业可以将内部知识库、文档系统通过MCP暴露给AI模型，实现智能问答和文档分析：
# 企业知识库MCP服务器示例
class EnterpriseKnowledgeServer(MCPServer):
"""企业知识库MCP服务器"""
def __init__(self):
super().__init__()
# 注册企业工具
self.register_tool(
"search_documents",
self.search_documents,
{
"name": "search_documents",
"description": "搜索企业内部文档",
"input_schema": {
"type": "object",
"properties": {
"query": {"type": "string"},
"department": {"type": "string", "enum": ["技术", "市场", "财务"]},
"date_range": {"type": "string"}
},
"required": ["query"]
}
}
)
self.register_tool(
"get_policy",
self.get_policy,
{
"name": "get_policy",
"description": "获取公司政策文档",
"input_schema": {
"type": "object",
"properties": {
"policy_type": {"type": "string"},
"version": {"type": "string"}
},
"required": ["policy_type"]
}
}
)
def search_documents(self, query, department=None, date_range=None):
"""搜索文档"""
# 连接企业文档系统
# 执行搜索
# 返回结果
return {
"results": [
{"title": "Q4报告", "summary": "...", "url": "..."},
{"title": "技术方案", "summary": "...", "url": "..."}
],
"count": 2
}
def get_policy(self, policy_type, version="latest"):
"""获取政策文档"""
# 从政策管理系统获取
return {
"title": f"{policy_type}政策",
"content": "...",
"effective_date": "2024-01-01"
}
# 使用示例
server = EnterpriseKnowledgeServer()
# AI模型可以调用search_documents、get_policy等工具
# 无需了解企业内部系统的具体实现细节2. 数据分析与可视化
MCP可以将数据分析工具、数据库查询、可视化库等封装为标准化工具：
# 数据分析MCP工具示例
data_analysis_tools = {
"query_database": {
"name": "query_database",
"description": "执行SQL查询",
"input_schema": {
"type": "object",
"properties": {
"sql": {"type": "string"},
"data_source": {"type": "string"}
},
"required": ["sql"]
}
},
"generate_chart": {
"name": "generate_chart",
"description": "生成数据图表",
"input_schema": {
"type": "object",
"properties": {
"data": {"type": "array"},
"chart_type": {"type": "string", "enum": ["line", "bar", "pie"]},
"title": {"type": "string"}
},
"required": ["data", "chart_type"]
}
},
"statistical_analysis": {
"name": "statistical_analysis",
"description": "执行统计分析",
"input_schema": {
"type": "object",
"properties": {
"data": {"type": "array"},
"analysis_type": {"type": "string", "enum": ["summary", "correlation", "regression"]}
},
"required": ["data", "analysis_type"]
}
}
}
# AI模型可以组合使用这些工具
# 1. 查询数据 → 2. 统计分析 → 3. 生成图表3. 自动化工作流
将业务流程、自动化脚本封装为MCP工具，实现AI驱动的自动化：
# 工作流自动化MCP示例
workflow_tools = {
"approve_request": {
"name": "approve_request",
"description": "审批请求",
"input_schema": {
"type": "object",
"properties": {
"request_id": {"type": "string"},
"action": {"type": "string", "enum": ["approve", "reject"]},
"comment": {"type": "string"}
},
"required": ["request_id", "action"]
}
},
"send_notification": {
"name": "send_notification",
"description": "发送通知",
"input_schema": {
"type": "object",
"properties": {
"recipient": {"type": "string"},
"message": {"type": "string"},
"channel": {"type": "string", "enum": ["email", "slack", "teams"]}
},
"required": ["recipient", "message"]
}
},
"create_task": {
"name": "create_task",
"description": "创建任务",
"input_schema": {
"type": "object",
"properties": {
"title": {"type": "string"},
"assignee": {"type": "string"},
"due_date": {"type": "string"}
},
"required": ["title", "assignee"]
}
}
}
# AI可以理解自然语言指令，自动调用相应工具
# 例如："请审批ID为123的采购请求，并通知采购部门"
# → 调用approve_request → 调用send_notification4. 代码开发与调试
MCP可以将开发工具、版本控制、测试框架等集成到AI编程助手中：
# 开发工具MCP集成
dev_tools = {
"search_code": {
"name": "search_code",
"description": "搜索代码库",
"input_schema": {
"type": "object",
"properties": {
"pattern": {"type": "string"},
"file_type": {"type": "string"},
"repository": {"type": "string"}
},
"required": ["pattern"]
}
},
"run_tests": {
"name": "run_tests",
"description": "运行测试",
"input_schema": {
"type": "object",
"properties": {
"test_path": {"type": "string"},
"filter": {"type": "string"}
}
}
},
"debug_code": {
"name": "debug_code",
"description": "调试代码",
"input_schema": {
"type": "object",
"properties": {
"file": {"type": "string"},
"line": {"type": "number"}
},
"required": ["file"]
}
}
}
# AI编程助手可以：
# 1. 搜索相关代码示例
# 2. 运行测试验证代码
# 3. 调试问题代码5. 物联网(IoT)与硬件控制
MCP可以将物联网设备、传感器、执行器封装为标准工具：
# IoT设备MCP集成
iot_tools = {
"get_sensor_data": {
"name": "get_sensor_data",
"description": "获取传感器数据",
"input_schema": {
"type": "object",
"properties": {
"device_id": {"type": "string"},
"sensor_type": {"type": "string", "enum": ["temperature", "humidity", "motion"]}
},
"required": ["device_id", "sensor_type"]
}
},
"control_device": {
"name": "control_device",
"description": "控制设备",
"input_schema": {
"type": "object",
"properties": {
"device_id": {"type": "string"},
"action": {"type": "string"},
"value": {"type": ["number", "string", "boolean"]}
},
"required": ["device_id", "action"]
}
}
}
# AI可以监控和控制智能家居、工业设备等
# 例如："调高客厅温度到24度"
# → 调用control_device("thermostat", "set_temperature", 24)6. 跨模型工具共享
不同AI模型可以通过MCP共享同一套工具，避免重复开发：
# 工具共享示例
# 公司内部建立中央MCP工具服务器
central_tool_server = MCPServer()
# 注册公司所有标准工具
central_tool_server.register_tool("查询客户信息", query_customer)
central_tool_server.register_tool("创建订单", create_order)
central_tool_server.register_tool("生成报告", generate_report)
# 不同团队、不同模型都使用同一套工具
# 团队A的GPT应用
gpt_client = MCPClient(central_tool_server.url)
# 团队B的Claude应用
claude_client = MCPClient(central_tool_server.url)
# 团队C的本地模型应用
local_client = MCPClient(central_tool_server.url)
# 所有客户端都有一致的工具体验2.5 MCP技术优势和行业价值
1. 技术优势
标准化与互操作性
# 标准化带来的好处
# 不同工具、不同模型可以无缝集成
standard_tool_interface = {
"name": "standard_name",
"description": "标准描述",
"input_schema": "标准输入模式",
"output_schema": "标准输出模式"
}
# 开发者只需学习一次MCP
# 就可以为所有支持MCP的模型提供工具工具复用与生态建设
# 工具市场概念
class MCPToolMarket:
"""MCP工具市场"""
def browse_tools(self, category):
"""浏览工具"""
return [
{"name": "天气查询", "author": "A公司", "rating": 4.5},
{"name": "股票分析", "author": "B公司", "rating": 4.2},
{"name": "代码检查", "author": "C公司", "rating": 4.8}
]
def install_tool(self, tool_id):
"""安装工具"""
# 下载工具包
# 注册到本地MCP服务器
# 立即可以使用
# 开发者可以：
# 1. 发布自己开发的工具
# 2. 使用他人开发的工具
# 3. 组合多个工具构建复杂应用性能优化
# MCP的性能优化机制
class OptimizedMCPServer(MCPServer):
"""优化的MCP服务器"""
def __init__(self):
super().__init__()
self.tool_cache = {} # 工具结果缓存
self.connection_pool = {} # 连接池
def call_tool_with_cache(self, tool_name, params, ttl=300):
"""带缓存的工具调用"""
cache_key = f"{tool_name}:{hash(str(params))}"
if cache_key in self.tool_cache:
cached = self.tool_cache[cache_key]
if time.time() - cached["timestamp"] < ttl:
return cached["result"] # 返回缓存结果
# 执行工具
result = self.call_tool(tool_name, params)
# 更新缓存
self.tool_cache[cache_key] = {
"result": result,
"timestamp": time.time()
}
return result安全与权限控制
# MCP的安全机制
class SecureMCPServer(MCPServer):
"""安全的MCP服务器"""
def __init__(self):
super().__init__()
self.acl = AccessControlList() # 访问控制列表
def call_tool(self, tool_name, parameters, user_context):
"""带权限检查的工具调用"""
# 检查用户权限
if not self.acl.can_access(user_context["user_id"], tool_name):
raise PermissionError(f"用户无权访问工具: {tool_name}")
# 检查参数安全性
self.validate_parameters(tool_name, parameters)
# 执行工具
result = super().call_tool(tool_name, parameters)
# 审计日志
self.audit_log(user_context, tool_name, parameters, result)
return result
def validate_parameters(self, tool_name, parameters):
"""验证参数安全性"""
# 防止SQL注入
# 防止命令注入
# 防止路径遍历
# 等安全检查2. 行业价值
降低AI应用开发门槛
# 传统AI应用开发 vs MCP开发
传统方式：
1. 为每个模型适配工具接口 √
2. 处理不同的调用约定 √
3. 管理复杂的集成逻辑 √
4. 重复开发相似功能 √
MCP方式：
1. 实现一次MCP工具 √
2. 所有MCP兼容模型自动可用 √
3. 专注于业务逻辑，而不是集成细节 √
4. 复用现有的MCP工具生态 √加速企业AI转型
# 企业AI转型路径
企业现状：
- 有多个业务系统
- 有内部工具和API
- 想引入AI但集成困难
MCP解决方案：
1. 将现有系统包装为MCP工具
2. 建立企业MCP工具目录
3. AI模型通过MCP访问所有系统
4. 逐步增加AI能力，不影响现有系统
# 技术债少，迁移风险低促进AI工具生态
# 健康的工具生态
工具开发者 → 发布高质量工具 → 获得收益/声誉
↓ ↑
工具市场平台 ← 质量控制/评级/分发
↓ ↑
用户 ← 发现/安装/使用工具 ← 反馈/需求
# 良性循环：
# 1. 更多工具吸引更多用户
# 2. 更多用户吸引更多开发者
# 3. 竞争促使工具质量提高
# 4. 生态更加繁荣推动AI标准化进程
# 标准化的重要性
现状问题：
- 每个AI公司有自己的工具接口
- 开发者需要学习多个SDK
- 工具无法在不同平台间复用
- 生态碎片化
MCP的贡献：
1. 提出统一标准 √
2. 得到行业主要参与者支持 √
3. 建立参考实现 √
4. 推动标准化进程 √
# 类似HTTP之于Web，SQL之于数据库3. 经济价值
降低成本
# 成本对比分析
传统方式成本：
1. 为每个模型开发适配器：N × 适配器成本
2. 维护多个代码库：N × 维护成本
3. 培训多个技术栈：N × 培训成本
4. 集成测试复杂度：指数级增长
MCP方式成本：
1. 开发MCP工具：1 × 工具成本
2. 维护单个标准：1 × 维护成本
3. 培训MCP标准：1 × 培训成本
4. 标准化测试：线性增长
# 随着模型数量N增加，节省成本越多创造新商业模式
# MCP催生的新商业模式
1. 工具即服务(Tool-as-a-Service)
- 提供专业工具订阅
- 按使用量收费
2. 工具市场平台
- 工具交易抽成
- 高级功能收费
3. 企业MCP解决方案
- 定制化工具开发
- 系统集成服务
4. 工具质量认证
- 工具测试认证
- 安全审计服务4. 社会价值
促进AI普及
# 降低AI使用门槛
之前：
中小企业 → 高额AI开发成本 → 难以承受 → 无法使用AI
现在：
中小企业 → 使用现成MCP工具 → 低成本集成AI → 享受AI红利
# 让更多组织和个人受益于AI技术推动技术创新
# 专注创新，而非重复劳动
之前：
开发者时间 = 80%集成工作 + 20%创新工作
现在：
开发者时间 = 20%MCP集成 + 80%创新工作
# 释放开发者的创造力
# 加速AI技术创新三、课后练习题与答案
3.1 选择题
-
MCP是什么的缩写？
A) Model Control Protocol
B) Model Context Protocol
C) Machine Communication Protocol
D) Model Connection Protocol
答案：B。MCP代表Model Context Protocol（模型上下文协议）。
-
下列哪项不是MCP的核心组件？
A) 工具(Tools)
B) 资源(Resources)
C) 模型权重(Weights)
D) 提示模板(Prompt Templates)
答案：C。模型权重是模型本身的参数，不是MCP协议的组成部分。
-
与OpenAI的Function Calling相比，MCP的主要优势是什么？
A) 只能用于Anthropic的模型
B) 是专有协议，性能更好
C) 开放标准，支持多模型
D) 不需要工具定义
答案：C。MCP是开放标准，支持多种AI模型，而Function Calling是OpenAI的专有实现。
-
MCP不支持以下哪种传输协议？
A) HTTP
B) WebSocket
C) SMTP
D) stdio
答案：C。SMTP是邮件传输协议，不适合MCP的实时通信需求。
-
在企业应用中，MCP的主要价值不包括：
A) 统一工具接口
B) 降低集成成本
C) 强制使用特定AI模型
D) 促进工具复用
答案：C。MCP的设计目标是模型无关性，不会强制使用特定模型。
3.2 填空题
-
MCP协议允许AI模型通过标准化的方式与外部工具和资源进行交互。
-
在MCP中，工具表示可执行的功能，资源表示只读的数据源。
-
MCP支持三种通信模式：请求-响应模式、服务器推送模式和双向流模式。
-
与Function Calling每次请求都发送工具定义不同，MCP在连接建立时发现和注册工具。
-
MCP的安全性机制包括认证、授权、参数验证和审计日志。
3.3 简答题
-
简述MCP协议的设计目标和核心原则。
答：MCP（模型上下文协议）的设计目标是建立一个开放、标准化的协议，使不同的AI模型能够以统一的方式与外部工具、数据源和系统进行交互。其核心设计原则包括：
-
协议无关性：不依赖特定传输协议，支持HTTP、WebSocket、stdio等多种方式
-
双向通信：支持客户端请求和服务器主动推送
-
强类型定义：使用JSON Schema等标准明确定义工具接口
-
可扩展性：易于添加新类型的工具和功能
-
安全性：内置认证、授权、参数验证等安全机制
-
模型无关性：不绑定特定AI模型，支持多种模型使用
-
-
对比MCP和Function Calling在工具发现机制上的差异。
答：MCP和Function Calling在工具发现机制上的主要差异如下：
-
Function Calling：
-
工具定义需要在每个API请求中作为参数发送
-
模型每次都需要解析工具定义
-
不支持动态添加或删除工具
-
工具列表在客户端控制，服务器被动接收
-
-
MCP：
-
工具在连接建立时通过专门的消息进行注册
-
客户端可以缓存工具定义，提高效率
-
支持服务器主动推送工具更新（添加、删除、修改）
-
支持工具的动态发现和订阅机制
-
工具列表在服务器端管理，客户端可以按需订阅
-
MCP的工具发现机制更加灵活和高效，特别适合工具数量多或需要动态更新的场景。
-
-
解释MCP如何促进AI工具生态的发展。
答：MCP通过以下方式促进AI工具生态的发展：
-
标准化接口：统一的工具定义和调用方式，降低开发门槛
-
工具复用：一次开发，多模型使用，提高工具利用率
-
市场机制：支持建立工具市场，开发者可以发布和销售工具
-
质量反馈：用户评级和反馈机制促进工具质量提升
-
组合创新：标准化接口使得工具可以方便地组合使用
-
生态正循环：更多工具吸引更多用户，更多用户吸引更多开发者
MCP类似互联网的HTTP协议，为AI工具建立了"通用语言"，使得工具开发者、模型提供商、最终用户都能在统一的框架下协作，从而推动整个生态的繁荣发展。
-
3.4 实操题
题目：实现一个简单的MCP服务器，提供两个工具：计算器和天气查询（模拟）。要求包括工具注册、请求处理和响应返回。
要求：
-
使用Python实现
-
支持工具发现和工具调用
-
实现基本的错误处理
-
提供简单的命令行界面
参考答案：
"""
简单的MCP服务器实现
提供计算器和天气查询两个工具
"""
import json
import sys
from typing import Dict, Any, List, Optional
import math
from datetime import datetime
class SimpleMCPServer:
"""简单的MCP服务器实现"""
def __init__(self):
"""初始化服务器，注册可用工具"""
self.tools = {}
self.register_tools()
def register_tools(self):
"""注册所有可用工具"""
# 注册计算器工具
self.register_tool(
name="calculator",
description="执行数学计算",
input_schema={
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "数学表达式，如：3 + 4 * 2"
}
},
"required": ["expression"]
},
handler=self.handle_calculator
)
# 注册天气查询工具
self.register_tool(
name="weather",
description="查询天气信息",
input_schema={
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称"
},
"date": {
"type": "string",
"description": "日期，格式：YYYY-MM-DD，默认今天"
}
},
"required": ["city"]
},
handler=self.handle_weather
)
def register_tool(self, name: str, description: str, input_schema: Dict, handler):
"""注册一个工具"""
self.tools[name] = {
"name": name,
"description": description,
"input_schema": input_schema,
"handler": handler
}
def handle_request(self, request_str: str) -> str:
"""处理MCP请求"""
try:
request = json.loads(request_str)
request_type = request.get("type")
if request_type == "list_tools":
return self.handle_list_tools()
elif request_type == "call_tool":
return self.handle_call_tool(
request.get("tool"),
request.get("parameters", {})
)
else:
return self.create_error_response(
"invalid_request",
f"未知的请求类型: {request_type}"
)
except json.JSONDecodeError:
return self.create_error_response(
"invalid_json",
"请求不是有效的JSON格式"
)
except Exception as e:
return self.create_error_response(
"server_error",
f"服务器内部错误: {str(e)}"
)
def handle_list_tools(self) -> str:
"""处理工具列表请求"""
tool_list = []
for tool in self.tools.values():
tool_list.append({
"name": tool["name"],
"description": tool["description"],
"input_schema": tool["input_schema"]
})
response = {
"type": "tool_list",
"tools": tool_list
}
return json.dumps(response, ensure_ascii=False)
def handle_call_tool(self, tool_name: str, parameters: Dict) -> str:
"""处理工具调用请求"""
if tool_name not in self.tools:
return self.create_error_response(
"tool_not_found",
f"工具不存在: {tool_name}"
)
tool = self.tools[tool_name]
try:
# 调用工具处理器
result = tool["handler"](parameters)
response = {
"type": "tool_result",
"tool": tool_name,
"result": result
}
return json.dumps(response, ensure_ascii=False)
except Exception as e:
return self.create_error_response(
"tool_error",
f"工具执行错误: {str(e)}"
)
def handle_calculator(self, parameters: Dict) -> Dict:
"""计算器工具处理器"""
expression = parameters.get("expression", "")
if not expression:
raise ValueError("表达式不能为空")
# 安全评估表达式
# 注意：实际生产环境应使用更安全的方式，如ast.literal_eval
# 这里简化处理，仅用于演示
allowed_names = {
'abs': abs,
'round': round,
'pow': pow,
'max': max,
'min': min,
'sum': sum,
}
# 添加数学函数
for name in ['sqrt', 'sin', 'cos', 'tan', 'log', 'log10', 'exp']:
if hasattr(math, name):
allowed_names[name] = getattr(math, name)
try:
# 警告：eval在受限环境下使用
# 生产环境应使用更安全的方法
result = eval(expression, {"__builtins__": {}}, allowed_names)
return {
"expression": expression,
"result": result,
"success": True
}
except Exception as e:
raise ValueError(f"计算表达式失败: {str(e)}")
def handle_weather(self, parameters: Dict) -> Dict:
"""天气查询工具处理器（模拟）"""
city = parameters.get("city", "未知城市")
date = parameters.get("date", datetime.now().strftime("%Y-%m-%d"))
# 模拟天气数据
weather_data = {
"北京": {"temp": 15, "condition": "晴", "humidity": 40},
"上海": {"temp": 18, "condition": "多云", "humidity": 65},
"广州": {"temp": 25, "condition": "小雨", "humidity": 85},
"深圳": {"temp": 26, "condition": "阵雨", "humidity": 80},
}
if city in weather_data:
data = weather_data[city]
return {
"city": city,
"date": date,
"temperature": data["temp"],
"condition": data["condition"],
"humidity": data["humidity"],
"unit": "摄氏度"
}
else:
# 返回模拟数据
return {
"city": city,
"date": date,
"temperature": 20,
"condition": "晴",
"humidity": 50,
"unit": "摄氏度",
"note": "模拟数据"
}
def create_error_response(self, error_code: str, message: str) -> str:
"""创建错误响应"""
response = {
"type": "error",
"error": {
"code": error_code,
"message": message
}
}
return json.dumps(response, ensure_ascii=False)
class MCPClient:
"""简单的MCP客户端"""
def __init__(self, server):
self.server = server
self.available_tools = []
def discover_tools(self):
"""发现服务器上的可用工具"""
request = {
"type": "list_tools"
}
response_str = self.server.handle_request(json.dumps(request))
response = json.loads(response_str)
if response["type"] == "tool_list":
self.available_tools = response["tools"]
return self.available_tools
else:
raise Exception(f"发现工具失败: {response}")
def call_tool(self, tool_name: str, parameters: Dict) -> Any:
"""调用工具"""
request = {
"type": "call_tool",
"tool": tool_name,
"parameters": parameters
}
response_str = self.server.handle_request(json.dumps(request))
response = json.loads(response_str)
if response["type"] == "tool_result":
return response["result"]
elif response["type"] == "error":
raise Exception(f"工具调用错误: {response['error']['message']}")
else:
raise Exception(f"未知响应类型: {response['type']}")
def run_demo():
"""运行演示"""
print("=" * 50)
print("MCP服务器演示")
print("=" * 50)
# 创建服务器
server = SimpleMCPServer()
# 创建客户端
client = MCPClient(server)
# 发现可用工具
print("\n1. 发现可用工具:")
tools = client.discover_tools()
for tool in tools:
print(f" - {tool['name']}: {tool['description']}")
# 测试计算器工具
print("\n2. 测试计算器工具:")
try:
result = client.call_tool("calculator", {
"expression": "3 + 4 * 2"
})
print(f" 计算 3 + 4 * 2 = {result['result']}")
result = client.call_tool("calculator", {
"expression": "sqrt(16) + sin(0)"
})
print(f" 计算 sqrt(16) + sin(0) = {result['result']}")
except Exception as e:
print(f" 计算器工具错误: {e}")
# 测试天气查询工具
print("\n3. 测试天气查询工具:")
try:
cities = ["北京", "上海", "未知城市"]
for city in cities:
result = client.call_tool("weather", {
"city": city,
"date": "2024-01-15"
})
print(f" {city}天气: {result['temperature']}°C, {result['condition']}")
except Exception as e:
print(f" 天气查询错误: {e}")
# 测试错误情况
print("\n4. 测试错误处理:")
try:
# 调用不存在的工具
result = client.call_tool("nonexistent", {})
except Exception as e:
print(f" 预期错误: {e}")
try:
# 调用计算器但缺少必要参数
result = client.call_tool("calculator", {})
except Exception as e:
print(f" 参数错误: {e}")
print("\n" + "=" * 50)
print("演示完成")
print("=" * 50)
def interactive_mode():
"""交互式模式"""
server = SimpleMCPServer()
client = MCPClient(server)
print("MCP交互式客户端")
print("输入 'list' 查看工具")
print("输入 'calc <表达式>' 使用计算器")
print("输入 'weather <城市>' 查询天气")
print("输入 'exit' 退出")
while True:
try:
user_input = input("\n> ").strip()
if not user_input:
continue
if user_input.lower() == 'exit':
break
if user_input.lower() == 'list':
tools = client.discover_tools()
for tool in tools:
print(f"{tool['name']}: {tool['description']}")
continue
if user_input.lower().startswith('calc '):
expression = user_input[5:].strip()
try:
result = client.call_tool("calculator", {
"expression": expression
})
print(f"结果: {result['result']}")
except Exception as e:
print(f"错误: {e}")
continue
if user_input.lower().startswith('weather '):
city = user_input[8:].strip()
try:
result = client.call_tool("weather", {
"city": city
})
print(f"{city}天气: {result['temperature']}°C, {result['condition']}, 湿度{result['humidity']}%")
except Exception as e:
print(f"错误: {e}")
continue
print("未知命令，请输入 'list', 'calc <表达式>', 'weather <城市>' 或 'exit'")
except KeyboardInterrupt:
print("\n再见！")
break
except Exception as e:
print(f"错误: {e}")
if __name__ == "__main__":
print("请选择模式:")
print("1. 运行演示")
print("2. 交互式模式")
choice = input("输入选择 (1 或 2): ").strip()
if choice == "1":
run_demo()
elif choice == "2":
interactive_mode()
else:
print("无效选择，运行演示...")
run_demo()代码说明：
SimpleMCPServer类：实现了基本的MCP服务器功能
工具注册和管理
请求分发和处理
错误处理和响应生成
MCPClient类：简单的客户端实现
工具发现
工具调用
结果处理
工具实现：
计算器工具：支持基本数学运算
天气查询工具：模拟天气数据返回
运行模式：
演示模式：自动运行测试用例
交互式模式：命令行交互使用工具
安全考虑：
输入验证
错误处理
受限的eval执行环境
这个示例展示了MCP的核心概念，包括工具定义、注册、发现、调用和错误处理。虽然是一个简化版本，但包含了MCP的基本要素，可以帮助理解MCP的工作原理。
四、总结
模型上下文协议（MCP）代表了AI工具化、标准化的重要发展方向。在AI技术快速演进、应用场景不断扩展的今天，MCP为解决工具互操作性、降低集成成本、促进生态发展提供了切实可行的方案。
4.1 从技术角度看
MCP的开放性和标准化设计打破了模型与工具之间的壁垒，使开发者能够以统一的方式为不同AI模型提供能力扩展。与专有方案相比，MCP的模型无关性确保了技术的长期兼容性和投资保护价值。其支持的工具动态发现、服务器主动推送、多传输协议等特性，使其能够适应从简单工具调用到复杂企业集成的各种场景。
4.2 从行业影响看
MCP的推广将加速AI技术的普及和应用。中小型企业可以通过标准化的工具接口快速集成AI能力，而不必面对复杂的技术选型和集成工作。工具开发者可以专注于核心功能实现，通过工具市场获得更广泛的用户群体。模型提供商可以减少在工具适配上的重复投入，专注于模型本身的优化和创新。
4.3 从生态建设看
MCP有望成为AI领域的"HTTP协议"，建立工具开发者、模型提供商、最终用户之间的通用语言。这种标准化将催生丰富的工具生态，促进工具质量的提升，降低AI应用开发门槛，最终让更多组织和个人能够受益于人工智能技术。
当然，MCP作为一个新兴标准，仍面临普及推广、工具生态建设、性能优化等挑战。但随着Anthropic、Google等主要参与方的推动，以及开发者社区的积极参与，MCP有望在AI工具标准化领域发挥越来越重要的作用。对于AI从业者而言，理解并掌握MCP不仅有助于当前的技术选型和架构设计，更是在为未来的AI标准化时代做好准备。
在AI技术从"炫技"走向"实用"的关键时期，类似MCP这样的标准化工作显得尤为重要。它们不仅是技术规范，更是推动整个行业向前发展的基础设施。随着MCP等标准的成熟和普及，我们有望看到一个更加开放、互联、高效的AI工具生态，让AI技术真正成为推动社会进步的通用能力。
🌟 感谢您耐心阅读到这里！
🚀 技术成长没有捷径，但每一次的阅读、思考和实践，都在默默缩短您与成功的距离。
💡 如果本文对您有所启发，欢迎点赞👍、收藏📌、分享📤给更多需要的伙伴！
🗣️ 期待在评论区看到您的想法、疑问或建议，我会认真回复，让我们共同探讨、一起进步～
🔔 关注我，持续获取更多干货内容！
🤗 我们下篇文章见！
AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。
更多推荐
8
0- 0
已为社区贡献200条内容
所有评论(0)