导航菜单

  • 1.什么是MCP
  • 2.MCP架构
  • 3.MCP服务器
  • 4.MCP客户端
  • 5.版本控制
  • 6.连接MCP服务器
  • 7.SDKs
  • 8.Inspector
  • 9.规范
  • 10.架构
  • 11.协议
  • 12.生命周期
  • 13.工具
  • 14.资源
  • 15.提示
  • 16.日志
  • 17.进度
  • 18.传输
  • 19.补全
  • 20.引导
  • 21.采样
  • 22.任务
  • 23.取消
  • 24.Ping
  • 25.根
  • 26.分页
  • 27.授权
  • 28.初始化
  • 29.工具
  • 30.资源
  • 31.结构化输出
  • 32.提示词
  • 33.上下文
  • 34.StreamableHTTP
  • 35.参数补全
  • 36.引导
  • 37.采样
  • 38.LowLevel
  • 39.任务
  • 40.取消
  • 41.ping
  • 42.根
  • 43.分页
  • 44.授权
  • 45.授权
  • Keycloak
  • asyncio
  • contextlib
  • httpx
  • pathlib
  • pydantic
  • queue
  • starlette
  • subprocess
  • threading
  • uvicorn
  • JSON-RPC
  • z
  • 1.Ping
  • 2 ping_client.py
  • 3. ping_server.py
  • 4.执行流程
    • 4.1 整体流程
    • 4.2 服务端 ping_server.py 讲解
      • 4.2.1. 核心结构
      • 4.2.2. 关键点
      • 4.2.3. 为何需要 list_tools?
    • 4.3 客户端 ping_client.py 讲解
      • 4.3.1. 核心流程
      • 4.3.2. 关键点
    • 4.4 时序图
    • 4.5 消息格式(JSON-RPC)
    • 4.6 运行方式

1.Ping #

  • 什么是 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. 发送 ping

4.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. 终端 1:uv run ping_server.py(或由 MCP Inspector 启动)
  2. 终端 2:uv run ping_client.py(或通过 Inspector 的 Ping 标签页测试)

客户端会完成初始化、发送两次 ping,并打印成功信息。

← 上一节 40.取消 下一节 42.根 →

访问验证

请输入访问令牌

Token不正确,请重新输入