了解MCP如何连接客户端、服务器和llm
MCP 建立在灵活、可扩展的体系结构之上，可实现LLM应用程序和集成之间的无缝通信。本文档涵盖了核心体系结构组件和概念。
概述
MCP遵循客户端-服务器架构，其中:
Hosts 是启动连接的LLM应用程序
Clients 在主机应用程序内保持与服务器的1:1连接
Servers 向客户端提供上下文、工具和提示
核心组件
协议层
协议层处理消息成帧、请求/响应链接和高级通信模式。
TypeScript比较好理解协议层意思，因此贴代码如下
class Protocol<Request, Notification, Result> {
// Handle incoming requests
setRequestHandler<T>(schema: T, handler: (request: T, extra: RequestHandlerExtra) => Promise<Result>): void
// Handle incoming notifications
setNotificationHandler<T>(schema: T, handler: (notification: T) => Promise<void>): void
// Send requests and await responses
request<T>(request: Request, schema: T, options?: RequestOptions): Promise<T>
// Send one-way notifications
notification(notification: Notification): Promise<void>
}关键类别包括:
ProtocolClientServer
传输层
传输层处理客户端和服务器之间的实际通信。MCP支持多种传输机制:
Stdio运输
使用标准输入/输出进行通信
本地流程的理想选择
使用SSE传输的HTTP
将服务器发送的事件用于服务器到客户端的消息
用于客户端到服务器消息的HTTP POST
所有运输使用JSON-RPC2.0交换消息。具体消息格式可查看。
消息类型
MCP有这些主要类型的消息:
请求期待对方的回应:
interface Request {
method: string;
params?: { ... };
}2. 结果是对请求的成功响应:
interface Result {
[key: string]: unknown;
}3. 错误指示请求失败:
interface Error {
code: number;
message: string;
data?: unknown;
}4. 通知是不期望响应的单向消息:
interface Notification {
method: string;
params?: { ... };
}连接生命周期
1.初始化
客户端发送
initialize具有协议版本和功能的请求服务器以其协议版本和功能进行响应
客户端发送
initialized作为确认的通知正常的消息交换开始
2.消息交换
初始化后，支持以下模式:
请求-响应: 客户端或服务器发送请求，另一个响应
通知: 任何一方发送单向消息
3.终止
任何一方都可以终止连接:
清洁关机通过
close()传输断开
错误条件
错误处理
MCP定义这些标准错误代码:
enum ErrorCode {
// Standard JSON-RPC error codes
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[types.Resource]:
return [
types.Resource(
uri="example://resource",
name="Example Resource"
)
]
async def main():
async with stdio_server() as streams:
await app.run(
streams[0],
streams[1],
app.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())最佳实践
Type的选择
本地通信
对本地进程使用stdio传输
高效的同机通信
简单的流程管理
远程通信
将SSE用于需要HTTP兼容性的方案
考虑安全隐患，包括身份验证和授权
消息处理
请求处理
彻底验证输入
使用类型安全架构
优雅地处理错误
实现超时
进度报告
对长操作使用进度标记
逐步报告进度
包括已知的总进度
错误管理
使用适当的错误代码
包括有用的错误消息
清理错误上的资源
安全注意事项
运输安全
使用TLS进行远程连接
验证连接来源
在需要时实施身份验证
消息验证
验证所有传入消息
防止恶意代码输入
检查邮件大小限制
验证json-rpc格式
资源保护
实施访问控制
验证资源路径
监控资源使用情况
速率限制请求
错误处理
不要泄露敏感信息
日志安全相关错误
实施适当的清理
处理DoS方案
调试和监控
日志记录
记录协议事件
跟踪消息流
监控性能
记录错误
诊断
实施健康检查
监控连接状态
跟踪资源使用情况
配置文件性能
测试
测试不同的传输
验证错误处理
测试边界case
负载测试服务器
评论区