PTP MCP Server
A Model Context Protocol (MCP) server for monitoring and analyzing Precision Time Protocol (PTP) systems in OpenShift clusters.
🚀 Features
- PTP Configuration Analysis: Parse and validate PTP configurations from OpenShift
- Real-time Log Monitoring: Access linuxptp daemon logs with intelligent parsing
- Natural Language Queries: Ask questions about PTP status in plain English
- Health Monitoring: Comprehensive PTP system health checks
- Synchronization Analysis: Monitor sync status, offsets, and BMCA state
- Clock Hierarchy: Track grandmaster and clock hierarchy information
- ITU-T Compliance: Validate configurations against ITU-T G.8275.1 standards
📋 Prerequisites
- Python 3.8 or higher
- OpenShift CLI (
oc) installed and configured - Access to OpenShift cluster with PTP operator installed
- PTP namespace (
openshift-ptp) exists
🛠️ Installation
- Clone the repository:
git clone https://github.com/aneeshkp/ptp-mcp-server.git
cd ptp-mcp-server
- Install dependencies:
pip install -r requirements.txt
- Verify OpenShift access:
oc whoami
oc get namespace openshift-ptp
🧪 Quick Testing
Run the comprehensive test suite:
Expected output:
🔍 PTP MCP Server API Quick Test
==================================================
Tests Passed: 8/8
Success Rate: 100.0%
🎉 ALL TESTS PASSED! Your API is ready for agent integration.
📚 API Endpoints
1. Configuration API
from ptp_tools import PTPTools
tools = PTPTools()
result = await tools.get_ptp_config({"namespace": "openshift-ptp"})
2. Logs API
result = await tools.get_ptp_logs({"lines": 1000})
3. Search API
result = await tools.search_logs({"query": "dpll", "time_range": "last_hour"})
4. Health API
result = await tools.check_ptp_health({"check_config": True, "check_sync": True})
5. Natural Language API
result = await tools.query_ptp({"question": "What is the current grandmaster?"})
6. Grandmaster Status API
result = await tools.get_grandmaster_status({"detailed": True})
7. Sync Status API
result = await tools.analyze_sync_status({"include_offsets": True})
8. Clock Hierarchy API
result = await tools.get_clock_hierarchy({"include_ports": True})
🚀 Usage Examples
Basic Health Check
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())
Natural Language Query
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())
Log Analysis
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 Server
Start the MCP server for integration with MCP-compatible clients:
The server provides the following MCP tools:
get_ptp_config - Get PTP configurationget_ptp_logs - Get linuxptp daemon logssearch_logs - Search logs for patternsget_grandmaster_status - Get grandmaster infoanalyze_sync_status - Analyze sync statusget_clock_hierarchy - Get clock hierarchycheck_ptp_health - Comprehensive health checkquery_ptp - Natural language interface
- Average Response Time: 0.78s
- Fastest API: Configuration API (0.22s)
- Concurrent Operations: 4/4 successful in 2.45s
- Success Rate: 100% (8/8 endpoints)
🏗️ Architecture
ptp-mcp-server/
├── ptp_mcp_server.py # Main MCP server
├── ptp_config_parser.py # PTP configuration parser
├── ptp_log_parser.py # Linuxptp log parser
├── ptp_model.py # PTP data models
├── ptp_query_engine.py # Natural language query engine
├── ptp_tools.py # API endpoint implementations
├── quick_test.py # Quick test suite
├── performance_test.py # Performance benchmarking
└── requirements.txt # Python dependencies
🔍 PTP Concepts Supported
- BMCA (Best Master Clock Algorithm): Clock selection and hierarchy
- Clock Types: OC (Ordinary Clock), BC (Boundary Clock), TC (Transparent Clock)
- ITU-T G.8275.1: Profile compliance and validation
- Synchronization: Offset tracking, frequency adjustment, sync status
- Grandmaster: Primary time source identification and status
- Clock Class: Quality and traceability indicators
- Domain Numbers: PTP domain configuration (24-43 for ITU-T)
🧪 Testing
Run All Tests
python performance_test.py
Individual Component Testing
# Test configuration parser
python -c "from ptp_config_parser import PTPConfigParser; import asyncio; asyncio.run(PTPConfigParser().get_ptp_configs())"
# Test log parser
python -c "from ptp_log_parser import PTPLogParser; import asyncio; asyncio.run(PTPLogParser().get_ptp_logs())"
📖 Documentation
🤝 Contributing
- Fork the repository
- Create a feature branch (
git checkout -b feature/amazing-feature) - Commit your changes (
git commit -m 'Add amazing feature') - Push to the branch (
git push origin feature/amazing-feature) - Open a Pull Request
📄 License
This project is licensed under the MIT License - see the LICENSE file for details.
🙏 Acknowledgments
- OpenShift PTP Operator team
- Linuxptp project
- Model Context Protocol (MCP) community
📞 Support
For issues and questions:
Status: ✅ Production Ready
Last Updated: January 2025
Version: 1.0.0