1.Ping #
本节介绍如何基于 MCP 协议实现最基础的「心跳检测」功能 —— Ping。
Ping 通常用于检测客户端与服务端之间的连接是否正常、是否断开,以及配合自动重连等机制实现高可用连接管理。
文件说明
- ping_client.py:Ping 客户端示例,负责发起 ping 请求,判断服务端是否存活,并演示多次 ping。
- ping_server.py:Ping 服务端示例,内置处理 ping 请求,仅做保活返回,不包含业务逻辑。为方便用
@modelcontextprotocol/inspector插件检测服务端能力,还声明了空的list_tools能力。
实现要点
- MCP 协议自带 Ping 机制:客户端通过
session.send_ping()发送 ping,服务端框架会自动响应,无需自行实现 ping 处理器。 - 客户端:初始化 MCP 会话后直接可调用
send_ping()检查连接活性。服务端若无响应/连接断开,会抛出异常。 - 服务端:继承内置 MCP Server,仅需启动(如基于标准输入输出或其他协议),无需显式添加 ping 处理;只需确保有至少一个能力被声明(如
list_tools)即可被 MCP Inspector 识别。
实际应用场景
- 可用于长期运行的 Agent、Bot、前端等需要稳固连接的交互场景。
- 便于自动化测试、监控、链路保活等需求。
- 如果基于 WebSocket、子进程桥接等方式部署,也可采用同样模式。
2 ping_client.py #
ping_client.py
# MCP Ping 客户端示例
"""MCP Ping 客户端
演示如何向服务端发送 ping 请求,用于检测连接是否存活。
"""
# 导入异步IO库
import asyncio
# 导入操作系统相关库
import os
# 导入系统库
import sys
# 从 mcp 模块导入所需类和函数
from mcp import ClientSession, StdioServerParameters, stdio_client
# 获取当前文件所在的目录路径
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
# 拼接虚拟环境下的python解释器路径(根据操作系统区分)
VENV_PYTHON = os.path.join(
BASE_DIR, ".venv", "Scripts" if os.name == "nt" else "bin",
"python.exe" if os.name == "nt" else "python",
)
# 异步主函数
async def main() -> None:
# 构建服务端参数(指定启动方式及参数)
server_params = StdioServerParameters(
command=VENV_PYTHON,
args=[os.path.join(BASE_DIR, "ping_server.py")],
env={},
cwd=BASE_DIR,
)
# 启动子进程并通过 stdio 连接服务端
async with stdio_client(server_params) as (read, write):
# 创建客户端会话对象
async with ClientSession(read, write) as session:
# 初始化会话
await session.initialize()
# 输出会话初始化完成
print("会话已初始化")
# 发送 ping 请求(服务端返回 EmptyResult 表示连接正常)
print("\n发送 ping...")
await session.send_ping()
print("Ping 成功,连接正常")
# 再次 ping 演示
print("\n再次发送 ping...")
await session.send_ping()
print("Ping 成功")
# 演示完成
print("\nPing 功能演示完成")
# 判断当前是否主程序运行入口
if __name__ == "__main__":
# 如果支持 reconfigure,则设置输入输出编码为utf-8
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8")
sys.stdin.reconfigure(encoding="utf-8")
# 运行主异步函数
asyncio.run(main())
3. ping_server.py #
ping_server.py
# MCP Ping 服务端
"""
MCP Ping 服务端
Ping 是 MCP 的心跳/保活机制,用于检测连接是否存活。
客户端发送 ping 请求,服务端返回空结果表示连接正常。
注意:MCP Inspector 要求服务器至少声明一项能力,否则会显示
"The connected server does not support any MCP capabilities"。
因此注册了空的 list_tools,使 capabilities.tools 非空。
"""
# 导入系统库,用于设置标准输入输出的编码
import sys
# 导入anyio库,用于异步运行
import anyio
# 导入mcp.types模块并起别名types,用于类型定义
import mcp.types as types
# 导入Server类,用于创建MCP服务端实例
from mcp.server import Server
# 导入stdio_server,用于基于标准输入输出实现通信
from mcp.server.stdio import stdio_server
# 创建服务器实例,并命名为"ping-server"(内置 ping 处理器)
server = Server("ping-server")
# 注册list_tools处理器,为MCP Inspector声明 tools 能力
@server.list_tools()
async def handle_list_tools() -> types.ListToolsResult:
# 返回空工具列表,仅声明tools能力,使MCP Inspector识别服务器
return types.ListToolsResult(tools=[])
# 异步主函数,负责启动服务端并监听客户端连接
async def main() -> None:
# 使用stdio_server进行异步上下文管理,获取读写流
async with stdio_server() as (read_stream, write_stream):
# 运行服务器,传入读流、写流和初始化参数
await server.run(
read_stream,
write_stream,
server.create_initialization_options(),
)
# 判断当前是否主程序入口
if __name__ == "__main__":
# 如果支持reconfigure,则设置标准输出和输入的编码为utf-8
if hasattr(sys.stdout, "reconfigure"):
sys.stdout.reconfigure(encoding="utf-8")
sys.stdin.reconfigure(encoding="utf-8")
# 使用anyio运行异步main函数
anyio.run(main)
4.执行流程 #
4.1 整体流程 #
Ping 是 MCP 的心跳/保活机制,用于检测连接是否存活。客户端发送 ping 请求,服务端返回空结果表示连接正常。
4.2 服务端 ping_server.py 讲解 #
4.2.1. 核心结构 #
server = Server("ping-server") # 创建服务器实例
@server.list_tools() # 注册空工具列表(声明 capabilities)
async def handle_list_tools() -> ... # 供 MCP Inspector 识别4.2.2. 关键点 #
| 部分 | 说明 |
|---|---|
stdio_server() |
从 stdin/stdout 创建读写流,用于与子进程通信 |
server.run() |
主循环:读取 JSON-RPC 消息、分发到对应处理器、写回响应 |
create_initialization_options() |
生成 InitializeResult(含 capabilities),供客户端初始化 |
| Ping 处理 | 由 SDK 内置 _ping_handler 处理,直接返回 EmptyResult,无需显式注册 |
4.2.3. 为何需要 list_tools? #
MCP Inspector 会检查 InitializeResult.capabilities。若所有能力为 None,会显示 “The connected server does not support any MCP capabilities”。注册空的 list_tools 后,capabilities.tools 非空,Inspector 能正确识别服务器。
4.3 客户端 ping_client.py 讲解 #
4.3.1. 核心流程 #
# 1. 构建服务端参数(子进程启动方式)
server_params = StdioServerParameters(
command=VENV_PYTHON, # .venv 中的 Python
args=["ping_server.py"], # 服务端脚本
cwd=BASE_DIR,
)
# 2. 启动子进程并建立 stdio 连接
async with stdio_client(server_params) as (read, write):
async with ClientSession(read, write) as session:
await session.initialize() # 3. 初始化握手
await session.send_ping() # 4. 发送 ping4.3.2. 关键点 #
| 部分 | 说明 |
|---|---|
StdioServerParameters |
描述如何启动服务端子进程(命令、参数、工作目录等) |
stdio_client() |
启动子进程,返回与其 stdin/stdout 绑定的读写流 |
ClientSession |
封装 JSON-RPC 会话,负责请求/响应匹配 |
session.initialize() |
发送 initialize 请求,获取 InitializeResult,建立会话 |
session.send_ping() |
发送 ping 请求,收到 EmptyResult 即表示连接正常 |
4.4 时序图 #
sequenceDiagram
participant C as 客户端<br/>ping_client.py
participant S as 服务端<br/>ping_server.py
participant SDK as MCP SDK
Note over C,S: 1. 建立连接
C->>S: stdio_client 启动子进程 ping_server.py
S->>S: stdio_server() 绑定 stdin/stdout
S->>S: server.run() 进入主循环
Note over C,S: 2. 初始化握手
C->>C: ClientSession(read, write)
C->>S: initialize (ClientCapabilities)
S->>S: create_initialization_options()
S->>S: get_capabilities() → tools 等
S->>C: InitializeResult (capabilities, serverInfo)
C->>C: print("会话已初始化")
Note over C,S: 3. 第一次 Ping
C->>S: ping (JSON-RPC Request, method="ping")
S->>S: _ping_handler (内置)
S->>C: EmptyResult (JSON-RPC Response)
C->>C: print("Ping 成功,连接正常")
Note over C,S: 4. 第二次 Ping
C->>S: ping
S->>C: EmptyResult
C->>C: print("Ping 成功")
Note over C,S: 5. 结束
C->>C: 退出 stdio_client 上下文
S->>S: 子进程退出
4.5 消息格式(JSON-RPC) #
Ping 请求:
{"jsonrpc": "2.0", "id": 1, "method": "ping", "params": null}Ping 响应:
{"jsonrpc": "2.0", "id": 1, "result": {"_meta": null}}result 为 EmptyResult,即 {"_meta": null},表示连接正常。
4.6 运行方式 #
- 终端 1:
uv run ping_server.py(或由 MCP Inspector 启动) - 终端 2:
uv run ping_client.py(或通过 Inspector 的 Ping 标签页测试)
客户端会完成初始化、发送两次 ping,并打印成功信息。