---
title: "告警 API 调用示例 | OceanBase 文档中心"
description: "告警 API 调用示例 本节提供脚本示例供用户了解如何对接告警 API 及测试告警 API 。 调用 OCP 接口示例 以下脚本以"推送告警事件接口"为例，介绍如何调用该接口。 说明 推送告警事件是指将告警推送给 OCP ，让 OCP 来处理告警消息。 #!/usr/bin/env python # -*- codi…"
---
切换语言

- 中文站 - 简体中文
- International - English
- 日本站 - 日本語

文档反馈![](https://mdn.alipayobjects.com/huamei_22khvb/afts/img/A*lJTmRZ61jSUAAAAAAAAAAAAADiGDAQ/original) 云平台 OCPV 4.0.0 企业版

# 告警 API 调用示例

更新时间：2026-04-14 16:45:43

本节提供脚本示例供用户了解如何对接告警 API 及测试告警 API 。

## 调用 OCP 接口示例

以下脚本以"推送告警事件接口"为例，介绍如何调用该接口。

#### 说明

推送告警事件是指将告警推送给 OCP ，让 OCP 来处理告警消息。

```python
#!/usr/bin/env python
# -*- coding: utf-8 -*-

import base64
import requests
import json

data = {
    "alarmType": "your_alarm_type",
    "labels": {"key":"value"},
    "target":"alarm_target",
}

base64userpass = base64.b64encode('{0}:{1}'.format('username', 'password'))
headers = {
    'Content-Type': 'application/json',
    'Accept': 'application/json',
    "Authorization": "Basic %s" % base64userpass,
}

resp = requests.post(url='http://192.168.0.1:8080/api/v2/alarm/alarms', headers=headers, data=json.dumps(data))
jresp = json.loads(resp.text)
print(jresp)

```

## 对接 OCP 告警示例

OCP 产生告警后，告警消息通过告警通道发送出去，通道可配置 Shell 脚本和 Python 脚本。该示例描述如何将告警通过通道发送出去。`send_alarm` 函数通过 os.environ 从环境变量获取告警相关的变量值，支持的变量列表可参见 [OCP 告警模板变量](https://www.oceanbase.com/docs/enterprise-oceanbase-ocp-cn-10000000000775923)。

以下为 Python 脚本示例：

# 注意：脚本内容第一行必须是 shebang 指定执行的程序，只支持 python 和 bash

import json
import requests
import os
import sys

ACCESS_TOKEN = "Ding Talk Token"

def send_alarm():
    """
    可以通过 os.environ 从环境变量获取 告警相关的变量值。
    支持的变量列表参见 《用户指南/附录4 OCP 告警模板变量》。
    """
    message = os.environ['message']
    data = {
        "msgtype": "markdown",
        "markdown": {
            "text": message
        }
    }
    resp = requests.post("https://oapi.dingtalk.com/robot/send?access_token=" + ACCESS_TOKEN, json=data)
    resp.close()
    response = json.loads(resp.text)

    # 将返回结果输入到标准错误（stderr）或标准输出（stdout），会被用于验证告警是否发送成功，优先验证stderr。
    if not response['errcode'] == 0:
        sys.stderr.write(resp.text)
    else:
        print(resp.text)

def main():
    """
    若发送成功，返回值为 0
    若发送失败，错误输出到 stderr，返回值不为 0
    """
    try:
        send_alarm()
        return 0
    except Exception as e:
        sys.stderr.write(str(e))
        return 1

if __name__ == "__main__":
    sys.exit(main())

```

## 应用举例

如需测试告警链路是否正常即告警是否可以推送到第三方告警平台，除了可以：

- 在 OCP 触发告警：

     1. 暂时停止 OCP-Agent 来触发 Agent 不可用相关告警如 `exporter数量不足` 告警。
     2. 查看第三方平台是否收到该告警。
 - OCP 通道设置中有测试功能，可直接通过 **发送测试消息** 按钮进行测试。参考 [新建告警通道](https://www.oceanbase.com/docs/enterprise-oceanbase-ocp-cn-10000000000775905)。

还可以通过上述"推送 OCP 告警事件接口"来构造告警，并通过告警通道脚本来发送告警到第三方平台。

 上一篇 下一篇 ![有帮助](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*y6ocSqN8cqsAAAAAAAAAAAAAARQnAQ)![无帮助](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*BG9IQJyLHF8AAAAAAAAAAAAAARQnAQ)![反馈](https://gw.alipayobjects.com/mdn/ob_asset/afts/img/A*eTWdQKCRKHwAAAAAAAAAAAAAARQnAQ)[AI](https://www.oceanbase.com/obi) 咨询热线