Ptp MCP Server

🚀 PTP MCP Server

PTP MCP Server 是一个模型上下文协议（MCP）服务器，用于在 OpenShift 集群中监控和分析精确时间协议（PTP）系统。它能帮助用户更好地管理和维护 PTP 系统，确保时间同步的准确性和稳定性。

🚀 快速开始

前提条件

• Python 3.8 或更高版本
• 安装并配置好 OpenShift CLI (oc)
• 能够访问已安装 PTP 运算符的 OpenShift 集群
• 存在 PTP 命名空间 (openshift-ptp)

安装步骤

1. 克隆仓库：

   git clone https://github.com/aneeshkp/ptp-mcp-server.git
   cd ptp-mcp-server
   

2. 安装依赖项：

   pip install -r requirements.txt
   

3. 验证 OpenShift 访问权限：

   oc whoami
   oc get namespace openshift-ptp
   

快速测试

运行综合测试套件：

python quick_test.py

预期输出：

🔍 PTP MCP Server API Quick Test
==================================================
Tests Passed: 8/8
Success Rate: 100.0%
🎉 ALL TESTS PASSED! Your API is ready for agent integration.

启动 MCP 服务器

启动 MCP 服务器以与支持 MCP 的客户端集成：

python ptp_mcp_server.py

✨ 主要特性

• PTP 配置分析：解析并验证来自 OpenShift 的 PTP 配置。
• 实时日志监控：通过智能解析访问 linuxptp 守护进程日志。
• 自然语言查询：使用通俗易懂的英语询问有关 PTP 状态的问题。
• 健康监控：对 PTP 系统进行全面的健康检查。
• 同步分析：监控同步状态、偏移量和最佳主时钟算法（BMCA）状态。
• 时钟层次结构：跟踪主时钟和时钟层次结构信息。
• ITU-T 合规性：根据 ITU-T G.8275.1 标准验证配置。

💻 使用示例

基础用法

基本健康检查

import asyncio
from ptp_tools import PTPTools

async def check_health():
    tools = PTPTools()
    health = await tools.check_ptp_health({})
    
    if health["success"]:
        print(f"Status: {health['overall_status']}")
        for check_name, result in health["checks"].items():
            print(f"{check_name}: {result}")
    else:
        print(f"Error: {health.get('error')}")

asyncio.run(check_health())

自然语言查询

async def ask_question():
    tools = PTPTools()
    response = await tools.query_ptp({
        "question": "What is the current grandmaster?"
    })
    
    if response["success"]:
        print(f"Answer: {response['response']}")
    else:
        print(f"Error: {response.get('error')}")

asyncio.run(ask_question())

日志分析

async def analyze_logs():
    tools = PTPTools()
    
    # Get recent logs
    logs = await tools.get_ptp_logs({"lines": 500})
    
    # Search for specific events
    sync_loss = await tools.search_logs({"query": "sync loss"})
    clock_changes = await tools.search_logs({"query": "clockClass change"})
    
    print(f"Total logs: {logs['logs_count']}")
    print(f"Sync loss events: {sync_loss['matching_logs']}")
    print(f"Clock changes: {clock_changes['matching_logs']}")

asyncio.run(analyze_logs())

高级用法

服务器提供以下 MCP 工具：

• get_ptp_config - 获取 PTP 配置
• get_ptp_logs - 获取 linuxptp 守护进程日志
• search_logs - 搜索日志中的模式
• get_grandmaster_status - 获取主时钟信息
• analyze_sync_status - 分析同步状态
• get_clock_hierarchy - 获取时钟层次结构
• check_ptp_health - 全面的健康检查
• query_ptp - 自然语言接口

📚 详细文档

API 端点

1. 配置 API

from ptp_tools import PTPTools
tools = PTPTools()
result = await tools.get_ptp_config({"namespace": "openshift-ptp"})

2. 日志 API

result = await tools.get_ptp_logs({"lines": 1000})

3. 搜索 API

result = await tools.search_logs({"query": "dpll", "time_range": "last_hour"})

4. 健康 API

result = await tools.check_ptp_health({"check_config": True, "check_sync": True})

5. 自然语言 API

result = await tools.query_ptp({"question": "What is the current grandmaster?"})

6. 主时钟状态 API

result = await tools.get_grandmaster_status({"detailed": True})

7. 同步状态 API

result = await tools.analyze_sync_status({"include_offsets": True})

8. 时钟层次结构 API

result = await tools.get_clock_hierarchy({"include_ports": True})

🔧 技术细节

性能指标

• 平均响应时间：0.78 秒
• 最快的 API：配置 API（0.22 秒）
• 并发操作：在 2.45 秒内 4/4 成功
• 成功率：100%（8/8 个端点）

架构

ptp-mcp-server/
├── ptp_mcp_server.py      # 主 MCP 服务器
├── ptp_config_parser.py   # PTP 配置解析器
├── ptp_log_parser.py      # Linuxptp 日志解析器
├── ptp_model.py           # PTP 数据模型
├── ptp_query_engine.py    # 自然语言查询引擎
├── ptp_tools.py           # API 端点实现
├── quick_test.py          # 快速测试套件
├── performance_test.py    # 性能基准测试
└── requirements.txt       # Python 依赖项

支持的 PTP 概念

• BMCA（最佳主时钟算法）：时钟选择和层次结构
• 时钟类型：OC（普通时钟）、BC（边界时钟）、TC（透明时钟）
• ITU-T G.8275.1：配置文件合规性和验证
• 同步：偏移跟踪、频率调整、同步状态
• 主时钟：主要时间源识别和状态
• 时钟类别：质量和可追溯性指标
• 域编号：PTP 域配置（ITU-T 为 24 - 43）

🧪 测试

运行所有测试

python quick_test.py

性能测试

python performance_test.py

单个组件测试

# 测试配置解析器
python -c "from ptp_config_parser import PTPConfigParser; import asyncio; asyncio.run(PTPConfigParser().get_ptp_configs())"

# 测试日志解析器
python -c "from ptp_log_parser import PTPLogParser; import asyncio; asyncio.run(PTPLogParser().get_ptp_logs())"

文档

• 测试指南 - 全面的测试说明
• 代理集成指南 - 代理的集成示例
• 测试步骤 - 分步测试过程
• 测试结果 - 完整的测试结果

🤝 贡献

1. 分叉仓库。
2. 创建功能分支 (git checkout -b feature/amazing-feature)。
3. 提交更改 (git commit -m 'Add amazing feature')。
4. 推送到分支 (git push origin feature/amazing-feature)。
5. 打开拉取请求。

📄 许可证

本项目采用 MIT 许可证 - 有关详细信息，请参阅 LICENSE 文件。

🙏 致谢

• OpenShift PTP 运算符团队
• Linuxptp 项目
• 模型上下文协议（MCP）社区

📞 支持

如有问题，请：

• 在 GitHub 上创建问题。
• 查看 测试文档。
• 查看 代理集成指南。

状态：✅ 可用于生产
最后更新时间：2025 年 1 月
版本：1.0.0