forked from xiaohei/taiji-AI-PAD
266 lines
8.5 KiB
Python
266 lines
8.5 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
MCP Server 监控 API 使用示例
|
||
演示如何使用监控接口获取系统指标、统计信息、趋势数据和告警
|
||
"""
|
||
|
||
import requests
|
||
import json
|
||
from typing import Optional
|
||
|
||
# 配置
|
||
BASE_URL = "http://localhost:8002"
|
||
API_KEY = "test-api-key"
|
||
|
||
HEADERS = {
|
||
"Content-Type": "application/json",
|
||
"X-API-Key": API_KEY
|
||
}
|
||
|
||
|
||
def get_system_metrics() -> Optional[dict]:
|
||
"""获取系统性能指标"""
|
||
print("\n=== 1. 获取系统性能指标 ===")
|
||
|
||
response = requests.get(
|
||
f"{BASE_URL}/api/v1/monitoring/metrics",
|
||
headers=HEADERS
|
||
)
|
||
|
||
print(f"状态: {response.status_code}")
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
|
||
# 系统指标
|
||
system = data.get('system', {})
|
||
print("\n系统资源:")
|
||
print(f" CPU 使用率: {system.get('cpu_usage_percent', 0):.1f}%")
|
||
print(f" 内存使用率: {system.get('memory_usage_percent', 0):.1f}%")
|
||
print(f" 内存使用: {system.get('memory_used_mb', 0):.0f} MB / {system.get('memory_total_mb', 0):.0f} MB")
|
||
print(f" 磁盘使用率: {system.get('disk_usage_percent', 0):.1f}%")
|
||
print(f" 磁盘使用: {system.get('disk_used_gb', 0):.1f} GB / {system.get('disk_total_gb', 0):.1f} GB")
|
||
|
||
# 服务指标
|
||
services = data.get('services', {})
|
||
print("\n服务统计:")
|
||
print(f" 活跃 Agent: {services.get('active_agents', 0)}")
|
||
print(f" 24h 执行次数: {services.get('total_executions_24h', 0)}")
|
||
print(f" 成功率: {services.get('success_rate_percent', 0):.1f}%")
|
||
print(f" 平均执行时间: {services.get('avg_execution_time_ms', 0):.2f} ms")
|
||
print(f" 日活用户: {services.get('daily_active_users', 0)}")
|
||
|
||
# 计费信息
|
||
billing = data.get('billing', {})
|
||
print("\n计费统计:")
|
||
print(f" 24h EU 消耗: {billing.get('total_eu_consumed_24h', 0):.2f}")
|
||
print(f" 24h 费用: ${billing.get('total_cost_24h', 0):.2f}")
|
||
|
||
return data
|
||
else:
|
||
print(f"获取失败: {response.text}")
|
||
return None
|
||
|
||
|
||
def get_service_stats(service: str = "all") -> Optional[dict]:
|
||
"""获取服务统计信息"""
|
||
print(f"\n=== 2. 获取服务统计 (service={service}) ===")
|
||
|
||
response = requests.get(
|
||
f"{BASE_URL}/api/v1/monitoring/stats?service={service}",
|
||
headers=HEADERS
|
||
)
|
||
|
||
print(f"状态: {response.status_code}")
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
stats = data.get('stats', {})
|
||
|
||
# Agent 统计
|
||
if 'agents' in stats:
|
||
agents = stats['agents']
|
||
print("\nAgent 统计:")
|
||
print(f" 总数: {agents.get('total', 0)}")
|
||
print(f" 活跃: {agents.get('active', 0)}")
|
||
print(f" 非活跃: {agents.get('inactive', 0)}")
|
||
print(f" 平均执行次数: {agents.get('avg_executions', 0):.1f}")
|
||
print(f" 平均成功率: {agents.get('avg_success_rate', 0):.1f}%")
|
||
|
||
# 执行统计
|
||
if 'executions' in stats:
|
||
executions = stats['executions']
|
||
print("\n执行统计 (7天):")
|
||
print(f" 总执行次数: {executions.get('total_7d', 0)}")
|
||
print(f" 已完成: {executions.get('completed', 0)}")
|
||
print(f" 失败: {executions.get('failed', 0)}")
|
||
print(f" 运行中: {executions.get('running', 0)}")
|
||
|
||
return data
|
||
else:
|
||
print(f"获取失败: {response.text}")
|
||
return None
|
||
|
||
|
||
def get_performance_trends(
|
||
metric: str = "executions",
|
||
period: str = "24h",
|
||
interval: str = "1h"
|
||
) -> Optional[dict]:
|
||
"""获取性能趋势数据"""
|
||
print(f"\n=== 3. 获取性能趋势 (metric={metric}, period={period}, interval={interval}) ===")
|
||
|
||
response = requests.get(
|
||
f"{BASE_URL}/api/v1/monitoring/trends?metric={metric}&period={period}&interval={interval}",
|
||
headers=HEADERS
|
||
)
|
||
|
||
print(f"状态: {response.status_code}")
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
|
||
print(f"\n指标: {data.get('metric')}")
|
||
print(f"周期: {data.get('period')}")
|
||
print(f"间隔: {data.get('interval')}")
|
||
|
||
trend_data = data.get('data', [])
|
||
print(f"\n数据点数量: {len(trend_data)}")
|
||
|
||
# 显示最近几个数据点
|
||
if trend_data:
|
||
print("\n最近的趋势数据:")
|
||
for item in trend_data[:5]: # 只显示前5个
|
||
print(f" {item.get('timestamp')}: 次数={item.get('count', 0)}, "
|
||
f"平均时间={item.get('avg_time_ms', 0):.2f}ms, "
|
||
f"成功率={item.get('success_rate', 0):.1f}%")
|
||
|
||
return data
|
||
else:
|
||
print(f"获取失败: {response.text}")
|
||
return None
|
||
|
||
|
||
def get_system_alerts(severity: Optional[str] = None) -> Optional[dict]:
|
||
"""获取系统告警"""
|
||
print(f"\n=== 4. 获取系统告警 (severity={severity or 'all'}) ===")
|
||
|
||
url = f"{BASE_URL}/api/v1/monitoring/alerts"
|
||
if severity:
|
||
url += f"?severity={severity}"
|
||
|
||
response = requests.get(url, headers=HEADERS)
|
||
|
||
print(f"状态: {response.status_code}")
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
alerts = data.get('alerts', [])
|
||
|
||
print(f"\n告警数量: {data.get('count', 0)}")
|
||
|
||
if alerts:
|
||
print("\n当前告警:")
|
||
for alert in alerts:
|
||
severity_icon = {
|
||
'info': 'ℹ️',
|
||
'warning': '⚠️',
|
||
'critical': '🔥'
|
||
}.get(alert.get('severity'), '•')
|
||
|
||
print(f" {severity_icon} [{alert.get('severity', 'unknown').upper()}] {alert.get('type')}")
|
||
print(f" {alert.get('message')}")
|
||
print(f" 时间: {alert.get('timestamp')}")
|
||
else:
|
||
print("\n✓ 无告警")
|
||
|
||
return data
|
||
else:
|
||
print(f"获取失败: {response.text}")
|
||
return None
|
||
|
||
|
||
def get_monitoring_dashboard() -> Optional[dict]:
|
||
"""获取监控仪表盘聚合数据"""
|
||
print("\n=== 5. 获取监控仪表盘 ===")
|
||
|
||
response = requests.get(
|
||
f"{BASE_URL}/api/v1/monitoring/dashboard",
|
||
headers=HEADERS
|
||
)
|
||
|
||
print(f"状态: {response.status_code}")
|
||
|
||
if response.status_code == 200:
|
||
data = response.json()
|
||
|
||
# 健康状态
|
||
health = data.get('health', {})
|
||
print(f"\n健康状态: {health.get('status', 'unknown')}")
|
||
|
||
# 告警摘要
|
||
alerts = data.get('alerts', {})
|
||
total_alerts = alerts.get('count', 0)
|
||
critical = alerts.get('critical_count', 0)
|
||
warning = alerts.get('warning_count', 0)
|
||
|
||
print(f"\n告警摘要:")
|
||
print(f" 总数: {total_alerts}")
|
||
print(f" 严重: {critical}")
|
||
print(f" 警告: {warning}")
|
||
|
||
# 详细数据在 metrics 和 stats 中
|
||
print("\n完整仪表盘数据已获取，包含:")
|
||
print(" - 系统指标 (CPU, 内存, 磁盘)")
|
||
print(" - 服务统计 (Agent, 执行, 用户)")
|
||
print(" - 告警详情")
|
||
|
||
return data
|
||
else:
|
||
print(f"获取失败: {response.text}")
|
||
return None
|
||
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("=" * 60)
|
||
print("MCP Server 监控 API 示例")
|
||
print("=" * 60)
|
||
|
||
try:
|
||
# 1. 获取系统指标
|
||
get_system_metrics()
|
||
|
||
# 2. 获取服务统计
|
||
get_service_stats("all")
|
||
get_service_stats("agents")
|
||
get_service_stats("executions")
|
||
|
||
# 3. 获取性能趋势
|
||
get_performance_trends("executions", "24h", "1h")
|
||
get_performance_trends("eu_consumption", "7d", "1d")
|
||
|
||
# 4. 获取告警
|
||
get_system_alerts()
|
||
get_system_alerts("warning")
|
||
get_system_alerts("critical")
|
||
|
||
# 5. 获取仪表盘聚合
|
||
get_monitoring_dashboard()
|
||
|
||
print("\n" + "=" * 60)
|
||
print("所有监控 API 示例执行完成！")
|
||
print("=" * 60)
|
||
|
||
except requests.exceptions.ConnectionError:
|
||
print("\n❌ 错误: 无法连接到 MCP Server")
|
||
print("请确保服务正在运行: docker-compose up -d mcp-server")
|
||
except Exception as e:
|
||
print(f"\n❌ 发生错误: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|