MCP协议-焦点架构
焦点架构了解MCP怎样毗连客户端、服务端和LLM
模型上下文协议（MCP）基于一个机动、可扩展的架构，使LLM应用步伐和集成之间的通讯无缝衔接。本文档涵盖了焦点架构组件和概念。
概述
MCP遵循客户端-服务端架构，此中：
[*]主机是启动毗连的LLM应用步伐（如Claude桌面或IDE）
[*]客户端与服务端保持1:1毗连，位于主机应用步伐内部
[*]服务端向客户端提供上下文、工具和提示
焦点组件
协议层
协议层处置惩罚消息框架、哀求/响应链接和高级通讯模式。
Python 示例代码
class Session(BaseSession):
async def send_request(
self,
request: RequestT,
result_type: type
) -> Result:
"""
Send request and wait for response. Raises McpError if response contains error.
"""
# Request handling implementation
async def send_notification(
self,
notification: NotificationT
) -> None:
"""Send one-way notification that doesn't expect response."""
# Notification handling implementation
async def _received_request(
self,
responder: RequestResponder
) -> None:
"""Handle incoming request from other side."""
# Request handling implementation
async def _received_notification(
self,
notification: ReceiveNotificationT
) -> None:
"""Handle incoming notification from other side."""
# Notification handling implementation
关键类包括：
[*]Protocol
[*]Client
[*]Server
传输层
传输层处置惩罚客户端和服务端之间的现实通讯。MCP支持多种传输机制：
[*] Stdio传输
[*]利用标准输入/输出进行通讯
[*]适用于本地进程
[*] HTTP与SSE传输
[*]利用服务端发送事件（SSE）进行服务端到客户端的消息传递
[*]利用HTTP POST进行客户端到服务端的消息传递
全部传输都利用JSON-RPC 2.0来互换消息。有关模型上下文协议消息格式的详细信息，请参阅规范。
消息范例
MCP有以下主要范例的消息：
[*]哀求期望从另一方获得响应：
interface Request {
method: string;
params?: { ... };
}
[*]效果是对哀求的成功响应：
interface Result {
: unknown;
}
[*]错误表现哀求失败：
interface Error {
code: number;
message: string;
data?: unknown;
}
[*]关照是单向消息，不期望响应：
interface Notification {
method: string;
params?: { ... };
}
毗连生命周期
1. 初始化
https://i-blog.csdnimg.cn/direct/2034569fa6bd4c0abbf5c46ccc494f22.png#pic_center
1.客户端发送`initialize`请求，包含协议版本和能力
2.服务端响应其协议版本和能力
3.客户端发送`initialized`通知作为确认
4.正常消息交换开始
2. 消息互换
初始化后，支持以下模式：
[*]哀求-响应：客户端或服务端发送哀求，另一方响应
[*]关照：任一方发送单向消息
3. 终止
任一方都可以终止毗连：
[*]通过close()进行干净关闭
[*]传输断开
[*]错误条件
错误处置惩罚
MCP定义了这些标准错误代码：
enum ErrorCode {
// 标准JSON-RPC错误代码
ParseError = -32700,
InvalidRequest = -32600,
MethodNotFound = -32601,
InvalidParams = -32602,
InternalError = -32603
}
SDK和应用步伐可以在-32000以上定义自己的错误代码。
错误通过以下方式传播：
[*]对哀求的错误响应
[*]传输上的错误事件
[*]协议级错误处置惩罚步伐
实现示例
以下是一个实现MCP服务端的根本示例：
Python 示例
import asyncio
import mcp.types as types
from mcp.server import Server
from mcp.server.stdio import stdio_server
app = Server("example-server")
@app.list_resources()
async def list_resources() -> list:
return [
types.Resource(
uri="example://resource",
name="Example Resource"
)
]
async def main():
async with stdio_server() as streams:
await app.run(
streams,
streams,
app.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main)
最佳实践
传输选择
[*] 本地通讯
[*]对本地进程利用stdio传输
[*]适用于同一台机器的通讯
[*]简单的进程管理
[*] 远程通讯
[*]对必要HTTP兼容性的场景利用SSE
[*]考虑安全影响，包括身份验证和授权
消息处置惩罚
[*] 哀求处置惩罚
[*]彻底验证输入
[*]利用范例安全的模式
[*]优雅地处置惩罚错误
[*]实现超时
[*] 进度报告
[*]对长时间操纵利用进度令牌
[*]逐步报告进度
[*]在已知时包括总进度
[*] 错误管理
[*]利用适当的错误代码
[*]包括有用的错误消息
[*]在错误时清理资源
安全注意事项
[*] 传输安全
[*]对远程毗连利用TLS
[*]验证毗连来源
[*]在必要时实现身份验证
[*] 消息验证
[*]验证全部传入消息
[*]清理输入
[*]检查消息巨细限制
[*]验证JSON-RPC格式
[*] 资源掩护
[*]实现访问控制
[*]验证资源路径
[*]监控资源利用环境
[*]限制哀求速率
[*] 错误处置惩罚
[*]不要泄露敏感信息
[*]记载与安全相关的错误
[*]实现适当的清理
[*]处置惩罚DoS场景
调试和监控
[*] 日志记载
[*]记载协议事件
[*]跟踪消息流
[*]监控性能
[*]记载错误
[*] 诊断
[*]实现健康检查
[*]监控毗连状态
[*]跟踪资源利用环境
[*]分析性能
[*] 测试
[*]测试不同的传输
[*]验证错误处置惩罚
[*]检查边沿环境
[*]负载测试服务端
免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。
页:
[1]