Skip to content

关于本页

本页介绍更夫所基于的开源内核 Keep 的通用能力。更夫在其上提供国内化增强:钉钉/企业微信/飞书通知、国产大模型接入、LDAP 认证与值班管理。部署与上手请看快速上手。

Provider 方法是 Provider 公开的除基本 query 和 notify 功能之外的其他功能(了解更多)。这些方法允许您以更具体的方式与 Provider 的 API 进行交互,实现更丰富的集成和自动化功能。

什么是 Provider 方法? ​

开发者使用 Provider 类中的 PROVIDER_METHODS 列表定义 Provider 方法。它们表示您可以通过 Provider 的 API 执行的特定操作或查询。这些方法扩展了 Provider 的基本功能,不仅限于简单的通知和查询。

例如,监控服务 Provider 可能会公开以下方法:

  • 静音/取消静音告警
  • 获取详细跟踪信息
  • 搜索特定指标
  • 修改监控配置

使用 Provider 方法 ​

您可以通过以下方式访问 Provider 方法:

  • 通过告警操作菜单访问 Keep 的平台界面
  • Keep 的智能 AI 助手(例如,"获取此告警的跟踪信息")
  • Keep 的 API
  • Keep 的工作流

通过 UI ​

当所选告警的源 Provider 可用时,方法会显示在告警操作菜单中:

::: note 如果参数在告警中可用,表单会自动填充方法所需的参数。 :::

通过 AI 助手 ​

Keep 的 AI 助手可以通过理解多个上下文来自动发现和调用 Provider 方法:

  1. 告警上下文:AI 理解:

    • 告警的源 Provider
    • 告警元数据和属性
    • 相关的服务和应用程序
    • 当前告警状态和严重性
  2. Provider 上下文:AI 知道:

    • 您已连接到账户的 Provider
    • 每个 Provider 的可用方法
    • 所需的参数及其类型
    • 方法描述和功能
  3. 历史上下文:AI 从以下内容中学习:

    • 类似的过去事件
    • 以前成功的方法调用
    • 告警解决中的常见模式

例如:

text
用户:您能获取此告警的跟踪信息吗?
助手:我看到此告警来自 Datadog。我将使用 Datadog Provider 的
get_traces 方法来获取跟踪信息。我将使用告警
元数据中的 trace_id:abc-123...

用户:此告警似乎与高延迟相关。您能帮助调查吗?
助手:我将帮助调查延迟问题。由于这是 Datadog 告警,
我可以:
1. 使用 search_traces() 获取最近的跟踪信息来查找慢请求
2. 使用 get_metrics() 获取指标来检查系统性能
3. 使用 search_logs() 查找相关日志

您想从我开始吗?

AI 助手自动执行以下操作:

  1. 识别相关的 Provider 方法
  2. 从上下文中提取所需参数
  3. 根据告警类型建议适当的操作
  4. 链接多个方法进行综合调查

通过 API ​

python
# Example using a Datadog provider method to mute a monitor
response = await api.post(
    f"/providers/{provider_id}/invoke/mute_monitor",
    {"monitor_id": "abc123", "duration": 3600}
)

添加新的 Provider 方法 ​

要为您的 Provider 添加新方法:

  1. 在您的 Provider 类中定义该方法(必须是实例方法):
python
def get_traces(self, trace_id: str) -> dict:
    """Get trace details from the provider.

    Args:
        trace_id (str): The ID of the trace to retrieve

    Returns:
        dict: The trace details
    """
    # Implementation
    pass
  1. 将方法元数据添加到 PROVIDER_METHODS:
python
from keep.providers.models.provider_method import ProviderMethod

PROVIDER_METHODS = [
    ProviderMethod(
        name="Get Traces",
        description="Retrieve trace details",
        func_name="get_traces",
        type="view",  # 'view' or 'action'
        scopes=["traces:read"],  # Required provider scopes
        category="Observability",  # Optional category for grouping methods
    )
]

注意:func_params 字段由 Keep 通过方法签名反射自动填充,因此您不需要手动定义它。

注意

Provider 方法必须是 Provider 类的实例方法(不是静态方法或类方法)。方法签名会自动检查以生成 UI 表单和参数验证。

完整示例 ​

以下是具有自定义方法的 Provider 的完整示例:

python
class MonitoringProvider(BaseProvider):
    PROVIDER_DISPLAY_NAME = "Monitoring Service"

    PROVIDER_METHODS = [
        ProviderMethod(
            name="Mute Alert",
            description="Mute an alert for a specified duration",
            func_name="mute_alert",
            type="action",
            scopes=["alerts:write"],
            category="Alert Management",
        ),
        ProviderMethod(
            name="Get Metrics",
            description="Retrieve metrics for a service",
            func_name="get_metrics",
            type="view",
            scopes=["metrics:read"],
            category="Observability",
        ),
    ]

    def mute_alert(self, alert_id: str, duration_minutes: int = 60) -> dict:
        """
        Mute an alert for the specified duration.

        Args:
            alert_id: The ID of the alert to mute
            duration_minutes: Duration to mute in minutes (default: 60)

        Returns:
            dict: Confirmation of the mute action
        """
        # Implementation here
        response = self._api_call(f"/alerts/{alert_id}/mute",
                                 {"duration": duration_minutes})
        return {"success": True, "muted_until": response["muted_until"]}

    def get_metrics(self, service_name: str, metric_type: str,
                   time_range: str = "1h") -> list:
        """
        Get metrics for a specific service.

        Args:
            service_name: Name of the service
            metric_type: Type of metric (cpu, memory, latency, etc.)
            time_range: Time range for metrics (default: "1h")

        Returns:
            list: List of metric data points
        """
        # Implementation here
        return self._query(f"metrics.{metric_type}",
                          service=service_name,
                          range=time_range)

方法类型 ​

  • view:返回用于显示的数据(例如,获取跟踪信息、指标)
  • action:执行操作(例如,静音告警、创建工单)

参数类型 ​

Provider 方法支持的参数类型:

  • str:字符串输入字段
  • int:数字输入字段
  • float:小数输入字段
  • bool:布尔复选框
  • datetime:日期/时间选择器
  • dict:JSON 对象输入
  • list:数组/列表输入
  • Literal:带预定义值的下拉菜单
  • Optional[type]:指定类型的可选参数

具有不同参数类型的示例:

python
from typing import Optional, Literal
from datetime import datetime

def advanced_query(
    self,
    metric_name: str,                                    # Required string
    time_range: Literal["1h", "6h", "24h", "7d"] = "1h", # Dropdown with options
    include_metadata: bool = False,                       # Boolean checkbox
    limit: Optional[int] = None,                          # Optional integer
    start_time: Optional[datetime] = None,                # Optional datetime picker
) -> dict:
    """Query metrics with advanced filtering options."""
    # Implementation
    pass

自动发现 ​

Keep 自动检查 Provider 类以:

  1. 发现可用的方法
  2. 提取参数信息
  3. 生成 UI 组件
  4. 启用 AI 对方法功能的理解

最佳实践 ​

  1. 清晰的文档:为方法提供详细的文档字符串
  2. 类型提示:为参数使用 Python 类型提示
  3. 错误处理:返回清晰的错误消息
  4. 作用域:定义最小所需作用域
  5. 验证:在执行前验证参数

限制 ​

  • 目前仅支持同步方法
  • 支持的参数类型仅限于基本类型
  • 方法必须是 Provider 类的实例方法
  • 方法通过反射自动发现
  • Keep 根据类型提示验证参数类型

基于 MIT 协议开源的 Keep · 由更夫社区维护