Skip to content

teams

Utils to send notifications to Microsoft Teams channels via incoming webhooks.

send_teams(task, status, message, webhook_secret)

Send a notification to a Microsoft Teams channel via an incoming webhook.

Parameters:

Name Type Description Default
task str

The task/workflow name.

 required
status str

The task status (e.g. "Succeeded", "Failed").

 required
message str

The message body (supports Markdown formatting).

 required
webhook_secret str

The name of the Azure Key Vault secret containing the webhook URL.

 required

Returns:

Type Description
Response

requests.Response: The response from the Teams webhook.

Raises:

Type Description
ValueError

If any required field is empty.
HTTPError

If the webhook returns an error status.
Example 
from physical_operations_utils.notification_utils import send_teams

send_teams(
    task="my-workflow-task-123",
    status="Succeeded",
    message="✨ Everything went well!",
    webhook_secret="TeamsWebhook-MyChannel",
)
Source code in physical_operations_utils/notification_utils/teams.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def send_teams(
    task: str,
    status: str,
    message: str,
    webhook_secret: str,
) -> requests.Response:
    """
    Send a notification to a Microsoft Teams channel via an incoming webhook.

    Args:
        task (str): The task/workflow name.
        status (str): The task status (e.g. "Succeeded", "Failed").
        message (str): The message body (supports Markdown formatting).
        webhook_secret (str): The name of the Azure Key Vault secret containing the
            webhook URL.

    Returns:
        requests.Response: The response from the Teams webhook.

    Raises:
        ValueError: If any required field is empty.
        requests.HTTPError: If the webhook returns an error status.

    Example:
        ```python
        from physical_operations_utils.notification_utils import send_teams

        send_teams(
            task="my-workflow-task-123",
            status="Succeeded",
            message="✨ Everything went well!",
            webhook_secret="TeamsWebhook-MyChannel",
        )
        ```
    """
    if not task or not task.strip():
        raise ValueError("Task must not be empty.")
    if not status or not status.strip():
        raise ValueError("Status must not be empty.")
    if not message or not message.strip():
        raise ValueError("Message must not be empty.")

    webhook_url = get_secret(webhook_secret)

    time = datetime.now(ZoneInfo("Europe/Oslo")).strftime("%Y-%m-%d at %I:%M:%S %p %Z")

    headers = {"Content-Type": "application/json", "User-Agent": "tripwire/1.0"}

    body = {
        "task": task,
        "status": status,
        "time": time,
        "message": message,
    }

    response = requests.post(webhook_url, headers=headers, json=body, timeout=(10, 30))
    response.raise_for_status()

    logging.info("Teams message sent successfully: %s", task)
    return response