MCP架構：全面解析AI工具集成的未來標準
作者：大模型之路
MCP架構的工作原理是讓AI應用程序（主機）創建獨立的客戶端，這些客戶端維護與不同服務器的單獨連接，從而實現流暢和模塊化的交互。
什么是Model Context Protocol (MCP)?
Model Context Protocol (MCP) 是一種現代化的標準化協議，它讓大型語言模型(LLM)能夠無縫連接外部工具和真實世界數據源，為AI中心化應用提供了前所未有的能力、靈活性和可維護性。
MCP正在成為社交媒體上的熱門話題，但關于它的實際定義和價值的討論卻常常缺乏清晰度。本文將全面解析MCP架構，幫助您理解這一AI工具集成的未來標準。
MCP架構的核心組件
MCP采用客戶端-服務器架構，由三個主要部分組成：
- MCP主機(Host)：這是用戶直接交互的AI應用程序，如Visual Studio Code、Claude桌面版或任何AI工具。主機負責管理整體體驗并協調通信。
- MCP客戶端(Client)：每個MCP客戶端由主機創建，用于連接特定的MCP服務器。它們是一對一的關系——一個客戶端只與一個服務器通信。客戶端的職責是維護連接并從服務器獲取有用的上下文或數據供主機使用。
- MCP服務器(Server)：這是提供AI應用所需上下文或數據的程序或服務。例如處理本地文件的服務器，或像Sentry MCP服務器那樣提供監控數據的服務。
實際示例：當Visual Studio Code連接到Sentry MCP服務器時，它會創建一個專用的MCP客戶端來管理該連接。如果它還需要連接到另一個服務器（如本地文件系統服務器），則會啟動另一個專用的MCP客戶端。這樣，每個服務器都有自己獨立的客戶端，保持連接的清晰和組織。
簡而言之，MCP架構的工作原理是讓AI應用程序（主機）創建獨立的客戶端，這些客戶端維護與不同服務器的單獨連接，從而實現流暢和模塊化的交互。
深入解析每個組件
主機(Host)
主機是面向用戶的主要應用程序，是用戶交互的起點。
MCP主機的例子：
- Claude Desktop — Anthropic的桌面應用程序
- Visual Studio Code — 帶有MCP擴展
- Cursor — 增強AI的IDE
- 開發者構建的自定義AI應用
主機可以是AI驅動的聊天界面、生產力工具（如Vs Code或Claude Desktop），或增強AI的集成開發環境（IDE）如Cursor。主機管理用戶輸入、顯示響應，并通常指導用戶、LLM和任何外部數據源或工具之間的信息流。
客戶端(Client)
客戶端充當翻譯器和通信管理器。
它駐留在主機內部，將用戶意圖編碼為結構化協議消息，然后安全地傳輸到MCP服務器。客戶端確保每條消息都遵循MCP規范，處理連接，并管理傳輸——無論應用程序是完全在本地運行還是與遠程服務器資源通信。
關鍵特性：
- 一對一關系：每個客戶端只連接一個服務器
- 協議執行：確保所有消息遵循MCP規范
- 連接管理：處理傳輸、安全和錯誤恢復
- 消息翻譯：將主機意圖轉換為結構化MCP請求
import asyncio
import json
from typing import Dict, Any, List
from pathlib import Path
from fastmcp import MCPClient
class FileManagerClient:
"""
文件管理器MCP服務器的客戶端
為AI助手提供通過MCP協議執行文件操作的高級接口
"""
def __init__(self, server_command: List[str]):
self.client = MCPClient(server_command)
self.connected = False
async def connect(self):
"""連接到MCP服務器"""
try:
await self.client.connect()
self.connected = True
print("? 已連接到文件管理器MCP服務器")
# 列出可用工具
tools = await self.client.list_tools()
print(f"?? 可用工具: {[tool['name'] for tool in tools.get('tools', [])]}")
except Exception as e:
print(f"? 連接失敗: {e}")
self.connected = False
async def read_file(self, file_path: str) -> Dict[str, Any]:
"""通過MCP讀取文件"""
if not self.connected:
return {"error": "未連接到服務器"}
try:
result = await self.client.call_tool("read_file", {"file_path": file_path})
return result
except Exception as e:
return {"error": f"讀取文件失敗: {e}"}
async def write_file(self, file_path: str, content: str) -> Dict[str, Any]:
"""通過MCP寫入文件"""
if not self.connected:
return {"error": "未連接到服務器"}
try:
result = await self.client.call_tool("write_file", {
"file_path": file_path,
"content": content
})
return result
except Exception as e:
return {"error": f"寫入文件失敗: {e}"}服務器(Server)
服務器是集成和執行層，是AI應用的數字"后臺"。
它連接數據庫、API、文件系統、知識庫和專用工具，以獲取、計算或轉換信息。服務器處理來自客戶端的請求，根據需要與外部資源交互，并以結構化、機器可讀的上下文響應——使LLM能夠直接在其推理中使用檢索到的信息。
關鍵特性：
- 與數據庫、API、文件系統和專用工具連接
- 處理來自客戶端的請求并獲取所需數據
- 提供結構化、機器可讀的響應
- 通過工具、資源和提示暴露能力
import asyncio
from typing import Dict, Any
from fastmcp import FastMCP
class CalculatorMCPServer:
"""
簡單計算器MCP服務器
為AI助手提供基本算術運算作為MCP工具
"""
def __init__(self):
self.mcp = FastMCP("Calculator Server")
self._setup_tools()
def _setup_tools(self):
"""注冊計算器工具"""
@self.mcp.tool()
async def add(a: float, b: float) -> Dict[str, Any]:
"""
兩數相加
參數:
a: 第一個數
b: 第二個數
返回:
a和b的和
"""
try:
result = a + b
return {
"operation": "addition",
"inputs": {"a": a, "b": b},
"result": result,
"formatted": f"{a} + = {result}"
}
except Exception as e:
return {"error": f"加法失敗: {str(e)}"}
@self.mcp.tool()
async def solve_quadratic(a: float, b: float, c: float) -> Dict[str, Any]:
"""
解二次方程 ax2 + bx + c = 0
參數:
a: x2的系數
b: x的系數
c: 常數項
返回:
二次方程的解
"""
try:
import math
if a == 0:
return {"error": "不是二次方程(a不能為0)"}
discriminant = b**2 - 4*a*c
if discriminant < 0:
return {
"operation": "quadratic_equation",
"equation": f"{a}x2 + x + {c} = 0",
"discriminant": discriminant,
"solutions": "無實數解(復數根)",
"note": "判別式為負"
}
elif discriminant == 0:
solution = -b / (2*a)
return {
"operation": "quadratic_equation",
"equation": f"{a}x2 + x + {c} = 0",
"discriminant": discriminant,
"solutions": [solution],
"formatted": f"x = {solution} (重根)"
}
else:
sqrt_discriminant = math.sqrt(discriminant)
solution1 = (-b + sqrt_discriminant) / (2*a)
solution2 = (-b - sqrt_discriminant) / (2*a)
return {
"operation": "quadratic_equation",
"equation": f"{a}x2 + x + {c} = 0",
"discriminant": discriminant,
"solutions": [solution1, solution2],
"formatted": f"x = {solution1} 或 x = {solution2}"
}
except Exception as e:
return {"error": f"解二次方程失敗: {str(e)}"}
async def run(self):
"""運行計算器MCP服務器"""
print("?? 計算器MCP服務器啟動中...")
print("可用工具: add, multiply, calculate_percentage, solve_quadratic")
await self.mcp.run(transport="stdio")MCP市場
以下是值得關注的MCP市場：
- mcpmarket.com — 一個即插即用的MCP服務器目錄，包含GitHub、Figma、Notion、Databricks等工具。
- mcp.so — 一個不斷增長的社區構建MCP服務器的開放倉庫。發現一個，分叉它，構建你自己的。
- Cline的MCP市場 — 一個由GitHub驅動的開源MCP連接器中心，任何人都可以使用。
MCP組件如何通信
MCP中的通信建立在JSON-RPC 2.0之上，為請求、響應和通知提供了清晰、可預測的結構。
MCP支持兩種主要的傳輸機制：
- STDIO：用于本地集成，服務器與客戶端在同一設備上運行。
- HTTP + SSE (服務器發送事件)：用于遠程部署，允許主機和服務器解耦，可以在云端、本地或混合設置中運行。
MCP請求的生命周期
典型的工作流程如下：
- 用戶通過主機應用程序提交查詢。
- 主機中的客戶端編碼此請求，確保遵循MCP格式。
- 服務器接收請求，使用其集成的工具、資源或提示獲取必要的數據或執行操作（設計用于檢索增強或代理系統）。
- 服務器將響應（結構化和LLM友好）發送回客戶端。
- 主機將結果呈現給用戶，完成閉環。
MCP服務器原語：工具、資源和提示
MCP的強大之處在于服務器以三種方式提供能力：
1. 工具(Tools)：可執行函數和操作
工具是可執行函數，支持基于代理的工作流和副作用。
將工具視為POST端點——它們執行操作、進行更改或執行操作。
from fastmcp import FastMCP
from pydantic import BaseModel, Field
import asyncio
import os
from pathlib import Path
from typing import List, Dict, Any, Optional
class FileManagerServer:
"""
AI助手的安全文件管理服務器
功能:
- 沙盒文件操作
- 權限檢查
- 安全路徑驗證
- 文件類型限制
"""
def __init__(self, allowed_directories: List[str], max_file_size: int = 10_000_000):
self.allowed_directories = [Path(d).resolve() for d in allowed_directories]
self.max_file_size = max_file_size
self.mcp = FastMCP("File Manager Server")
self._setup_tools()
def _is_path_allowed(self, file_path: str) -> bool:
"""檢查文件路徑是否在允許的目錄中"""
try:
resolved_path = Path(file_path).resolve()
return any(
str(resolved_path).startswith(str(allowed_dir))
for allowed_dir in self.allowed_directories
)
except Exception:
return False
def _setup_tools(self):
"""注冊MCP工具"""
@self.mcp.tool()
async def read_file(file_path: str) -> Dict[str, Any]:
"""
安全讀取文件內容
參數:
file_path: 要讀取的文件路徑
返回:
包含文件內容和元數據的字典
"""
if not self._is_path_allowed(file_path):
return {"error": "訪問被拒絕: 路徑不在允許的目錄中"}
try:
path = Path(file_path)
if not path.exists():
return {"error": f"文件未找到: {file_path}"}
if path.stat().st_size > self.max_file_size:
return {"error": f"文件過大(最大 {self.max_file_size} 字節)"}
# 處理不同的文件類型
import mimetypes
mime_type, _ = mimetypes.guess_type(file_path)
if mime_type and mime_type.startswith('text/'):
with open(path, 'r', encoding='utf-8') as f:
content = f.read()
return {
"content": content,
"type": "text",
"mime_type": mime_type,
"size": len(content),
"path": file_path
}
else:
# 對于二進制文件，返回base64
import base64
with open(path, 'rb') as f:
content = base64.b64encode(f.read()).decode('utf-8')
return {
"content": content,
"type": "binary",
"encoding": "base64",
"mime_type": mime_type,
"size": path.stat().st_size,
"path": file_path
}
except Exception as e:
return {"error": f"讀取文件失敗: {str(e)}"}
@self.mcp.tool()
async def write_file(file_path: str, content: str, mode: str = "w") -> Dict[str, Any]:
"""
安全寫入文件內容
參數:
file_path: 要寫入的文件路徑
content: 要寫入的內容
mode: 寫入模式('w'表示文本，'wb'表示二進制)
返回:
成功狀態和文件信息
"""
if not self._is_path_allowed(file_path):
return {"error": "訪問被拒絕: 路徑不在允許的目錄中"}
try:
path = Path(file_path)
# 如果父目錄不存在則創建
path.parent.mkdir(parents=True, exist_ok=True)
# 檢查內容大小
if len(content.encode('utf-8')) > self.max_file_size:
return {"error": f"內容過大(最大 {self.max_file_size} 字節)"}
# 寫入文件
if mode == "wb":
# 處理base64二進制內容
import base64
binary_content = base64.b64decode(content)
with open(path, 'wb') as f:
f.write(binary_content)
else:
with open(path, 'w', encoding='utf-8') as f:
f.write(content)
return {
"success": True,
"message": f"文件寫入成功: {file_path}",
"size": path.stat().st_size
}
except Exception as e:
return {"error": f"寫入文件失敗: {str(e)}"}2. 資源(Resources)：只讀數據訪問
資源提供對文件、文檔、知識庫和實時數據的訪問，用于檢索增強生成(RAG)。
將資源視為GET端點——它們提供信息而沒有副作用。
import json
import mimetypes
from datetime import datetime
@self.mcp.tool()
async def list_directory(directory_path: str, include_hidden: bool = False) -> Dict[str, Any]:
"""
列出目錄內容
參數:
directory_path: 要列出的目錄路徑
include_hidden: 是否包含隱藏文件
返回:
包含元數據的文件和目錄列表
"""
if not self._is_path_allowed(directory_path):
return {"error": "訪問被拒絕: 路徑不在允許的目錄中"}
try:
path = Path(directory_path)
if not path.exists():
return {"error": f"目錄未找到: {directory_path}"}
if not path.is_dir():
return {"error": f"路徑不是目錄: {directory_path}"}
files = []
for item in path.iterdir():
if not include_hidden and item.name.startswith('.'):
continue
stat = item.stat()
mime_type, _ = mimetypes.guess_type(str(item))
file_info = {
"name": item.name,
"path": str(item),
"size": stat.st_size,
"modified": datetime.fromtimestamp(stat.st_mtime).isoformat(),
"is_directory": item.is_dir(),
"mime_type": mime_type
}
files.append(file_info)
return {
"directory": directory_path,
"files": files,
"count": len(files)
}
except Exception as e:
return {"error": f"列出目錄失敗: {str(e)}"}
@self.mcp.tool()
async def search_files(
directory_path: str,
pattern: str,
file_type: Optional[str] = None,
max_results: int = 100
) -> Dict[str, Any]:
"""
搜索匹配模式的文件
參數:
directory_path: 要搜索的目錄
pattern: 搜索模式(支持通配符)
file_type: 按文件擴展名過濾(如'.py', '.txt')
max_results: 最大結果數
返回:
包含文件信息的搜索結果
"""
if not self._is_path_allowed(directory_path):
return {"error": "訪問被拒絕: 路徑不在允許的目錄中"}
try:
import fnmatch
path = Path(directory_path)
if not path.exists() or not path.is_dir():
return {"error": f"無效目錄: {directory_path}"}
matching_files = []
for item in path.rglob('*'):
if len(matching_files) >= max_results:
break
if item.is_file():
# 檢查模式匹配
if fnmatch.fnmatch(item.name.lower(), pattern.lower()):
# 檢查文件類型過濾器
if file_type and not item.name.lower().endswith(file_type.lower()):
continue
stat = item.stat()
file_info = {
"name": item.name,
"path": str(item),
"size": stat.st_size,
"modified": datetime.fromtimestamp(stat.st_mtime).isoformat()
}
matching_files.append(file_info)
return {
"search_query": pattern,
"directory": directory_path,
"file_type_filter": file_type,
"results": matching_files,
"count": len(matching_files),
"truncated": len(matching_files) >= max_results
}
except Exception as e:
return {"error": f"搜索失敗: {str(e)}"}3. 提示(Prompts)：可重用交互模板
提示是預定義的指令模板，指導模型輸出并減少用戶提示工程的需求。
from mcp.types import Prompt, PromptArgument, PromptMessage
@app.list_prompts()
async def list_prompts() -> list[Prompt]:
"""定義可重用的提示模板"""
return [
Prompt(
name="code_review",
description="全面的代碼審查模板",
arguments=[
PromptArgument(
name="code",
description="要審查的代碼",
required=True
),
PromptArgument(
name="language",
description="編程語言",
required=False
),
PromptArgument(
name="focus_areas",
description="要關注的特定領域(安全、性能等)",
required=False
)
]
),
Prompt(
name="data_analysis",
description="數據分析和洞察模板",
arguments=[
PromptArgument(
name="dataset",
description="要分析的數據集",
required=True
),
PromptArgument(
name="analysis_type",
description="分析類型(描述性、預測性等)",
required=False
)
]
),
Prompt(
name="technical_documentation",
description="生成技術文檔",
arguments=[
PromptArgument(
name="topic",
description="要記錄的主題",
required=True
),
PromptArgument(
name="audience",
description="目標受眾(開發者、用戶等)",
required=False
),
PromptArgument(
name="format",
description="文檔格式(API、教程、參考)",
required=False
)
]
)
]
@app.get_prompt()
async def get_prompt(name: str, arguments: dict) -> PromptMessage:
"""根據模板和參數生成提示內容"""
if name == "code_review":
code = arguments["code"]
language = arguments.get("language", "未知")
focus_areas = arguments.get("focus_areas", "通用最佳實踐")
prompt_content = f"""
請對以下{language}代碼進行徹底的代碼審查。
關注領域: {focus_areas}
要審查的代碼:
```{language}
{code}```
為什么MCP很重要
- 靈活性：您可以在單個設備上運行所有內容（主機、客戶端和服務器）以確保隱私和速度，或者將服務器拆分到云環境中以實現可擴展性和高可用性。
- 模塊化：組件可以混合、匹配和獨立升級，為開發人員提供了對集成深度、更新周期和安全邊界的巨大控制。
- 可擴展性：構建一次，隨處使用——任何兼容的主機或服務器都可以互操作。
- 可維護性：標準化接口使升級、故障排除和入門更加容易。
- 供應商中立性和開放性：避免鎖定；根據需要混合和匹配組件。
- 速度：得益于成熟的構建塊，開發速度更快，無需為每個新工具或數據集定制集成。
責任編輯：武曉燕
來源：
大模型之路
