Custom Python, Golang, etc integration

Profundis uses Server Sent Events (SSE) to notify you when one of your alert tasks matches. This makes it extremely easy to integrate it into your own workflow and trigger actions as soon as you receive an alert.

warning

You can have a maximum of 1 client connected and listening to events. Be aware that the Profundis alerting Docker image connects using SSE and then cannot be used in parallel of custom scripts/tools.

info

To receive notifications from your alerts, you need an API token.

When developping a custom integration to the Profundis notification system, your code should handle the SSE reconnections (cf the heartbeat messages).

Bash

You can try to receive events using a simple curl command in a terminal.

API_KEY="xxx"

curl -N \
-H "Accept: text/event-stream" \
"https://api.profundis.io/api/v2/sse/common/alerting?api_key=$API_KEY"

You will then receive heartbeat messages as below and the hosts which match your alert tasks.

id: 1749147000
data: {"type":"heartbeat"}

id: 1749147028
data: {"type":"heartbeat"}

Golang

The following example uses the r3labs SSE library. This is just an example. Feel free to build your own SSE client.

package main

import (
  "fmt"
  "log"

  "github.com/r3labs/sse/v2"
)

const (
  apiKey = "YOUR-API-KEY-HERE"
  apiURL = "https://api.profundis.io/api/v2/sse/common/alerting"
)

func main() {
  client := sse.NewClient(apiURL + "?api_key=" + apiKey)

  err := client.Subscribe("messages", func(msg *sse.Event) {
    // TODO: Unmarshal JSON here to parse alert/heartbeat data
    
    fmt.Printf("Received: %s\n", string(msg.Data))
  })

  if err != nil {
    log.Fatal("Error subscribing to SSE:", err)
  }
}

Python

The following example uses Python and the official requests package.

import requests

API_KEY = "YOUR-API-KEY-HERE"
API_URL = "https://api.profundis.io/api/v2/sse/common/alerting"

def main():
    headers = {
        "Accept": "text/event-stream"
    }
    
    params = {
        "api_key": API_KEY
    }
    
    response = requests.get(API_URL, headers=headers, params=params, stream=True)
    
    if response.status_code != 200:
        print(f"HTTP error: {response.status_code}")
        return
    
    print("Connected to SSE stream")
    
    for line in response.iter_lines(decode_unicode=True):
        if line and line.startswith("data: "):
            data = line[6:]  # Remove "data: " prefix
            print(f"Received: {data}")
            
            # TODO: Parse JSON here to handle alert/heartbeat data
            # Example: json.loads(data)

if __name__ == "__main__":
    main()