将PingZen集成为MCP服务器:126个AI代理工具
MCP服务器与主应用程序在同一进程中运行,但拥有独立的端点。STDIO特别适合桌面代理通过标准输入/输出启动Python脚本。HTTP传输则内置于FastAPI中,适用于云端服务。
126个工具的注册通过SDK装饰器完成。工具按以下类别分组:
- 监控:
create_monitor、update_monitor、delete_monitor、get_monitor_status、list_monitors、pause_monitor、resume_monitor。 - 告警:
create_alert、update_alert、test_alert、list_alerts。 - 事件:
list_incidents、get_incident_details、resolve_incident、add_incident_update。 - 心跳:
create_heartbeat、send_heartbeat_ping、get_heartbeat_status。 - 状态页:
create_status_page、add_monitor_to_status_page、update_status_page。 - Webhook:
create_webhook、list_webhooks。 - 报告:
create_scheduled_report、get_report_history。 - API密钥:
create_api_key、revoke_api_key。
工具注册示例
from mcp.server import Server
import mcp.types as types
server = Server("pingzen-mcp")
@server.list_tools()
async def handle_list_tools() -> list[types.Tool]:
return [
types.Tool(
name="create_monitor",
description="创建新监控",
inputSchema={
"type": "object",
"properties": {
"name": {"type": "string"},
"type": {"type": "string", "enum": ["http", "tcp", "udp", "icmp", "dns", "smtp", "transaction", ...]},
"target": {"type": "string"},
"interval": {"type": "integer", "minimum": 60},
},
"required": ["name", "type", "target"]
}
),
]
@server.call_tool()
async def handle_call_tool(name: str, arguments: dict) -> list[types.TextContent]:
if name == "create_monitor":
monitor = await monitor_service.create(arguments)
return [types.TextContent(type="text", text=json.dumps(monitor.to_dict()))]
参数以JSON Schema形式传递,简化了如使用Playwright实现的事务协议等复杂协议的定义。
OAuth 2.0认证
遵循RFC 9728规范,OAuth 2.0实现零配置。代理通过浏览器登录Telegram、Google或Yandex完成认证。授权后自动发放令牌。在CI/CD流水线中,可通过环境变量获取API密钥。只需在代理设置中填写URL https://pingzen.dev/mcp即可。
实用工具使用场景
创建HTTP监控示例:
{
"name": "create_monitor",
"arguments": {
"name": "Habr",
"type": "http",
"target": "https://habr.com",
"interval": 60,
"alert_channels": ["telegram"]
}
}
返回结果:监控ID及确认信息。此流程同样适用于list_monitors、resolve_incident或间隔24小时、宽限期1小时的心跳检测。
实现细节要点
- 验证机制:使用JSON Schema强制类型校验。处理函数中会显式进行类型转换(如将间隔设为整数),即使AI返回的是字符串也适用。
- 同步性设计:所有工具均为同步调用并带超时机制,避免引入后台任务以保持简洁。
- 错误处理:统一错误信息定义于
errors.py中:
_ERROR_MESSAGES = {
"INVALID_PROTOCOL": "无效的监控类型 '{type}'。支持类型:http, tcp, udp, icmp, dns, smtp, transaction, ...",
"TIMEOUT": "请求在 {timeout} 秒后超时",
}
- 日志输出:通过stderr输出,包含调用参数,适用于STDIO模式。
核心优势总结
- 126个工具覆盖从监控创建到事件管理与报告的完整生命周期。
- 双传输模式(STDIO + HTTP)兼顾本地与云端代理的灵活性。
- 零配置OAuth大幅降低部署门槛;API密钥支持自动化流程。
- 同步设计结合超时与严格验证,确保行为可预测。
- 无缝集成CI/CD与开发环境,无需频繁切换上下文。
— Editorial Team
暂无评论。