Teams presence and telephony status monitor.

Covers: single/bulk presence queries, preferred presence, status messages, polling for state changes, and business-rules routing logic.

Permissions

Required permissions: Presence.Read, Presence.Read.All, Presence.ReadWrite.

View source

from __future__ import annotations

from datetime import datetime, timedelta, timezone
from time import sleep
from typing import Optional

from office365.communications.presences.presence import Presence
from office365.graph_client import GraphClient
from tests.settings import admin_username, client_id, tenant

POLL_INTERVAL = 30
BUSY_ACTIVITIES = {"InACall", "InAConferenceCall", "InAMeeting", "Presenting"}


def is_available(presence: Presence) -> bool:
    if not presence.availability or not presence.activity:
        return False
    if presence.availability in ("DoNotDisturb", "Offline", "PresenceUnknown"):
        return False
    if presence.activity in BUSY_ACTIVITIES | {"OffWork", "OutOfOffice"}:
        return False
    return True


def agent_status_summary(presence: Presence) -> str:
    a = presence.availability or "Unknown"
    b = presence.activity or "Unknown"
    return f"availability={a}, activity={b}"


def take_agent_offline(client: GraphClient, user_id: str):
    expires = datetime.now(timezone.utc) + timedelta(hours=9)
    client.communications.presences[user_id].set_user_preferred_presence(
        availability="DoNotDisturb",
        activity="OffWork",
        expiration_duration="PT9H",
    ).execute_query()
    client.communications.presences[user_id].set_status_message(
        message="Shift ended",
        expiry=expires,
    ).execute_query()


def start_shift(client: GraphClient, user_id: str):
    expires = datetime.now(timezone.utc) + timedelta(hours=9)
    client.communications.presences[user_id].set_user_preferred_presence(
        availability="Available",
        activity="Available",
        expiration_duration="PT9H",
    ).execute_query()
    client.communications.presences[user_id].set_status_message(
        message="Online and ready",
        expiry=expires,
    ).execute_query()


def find_available_agent(client: GraphClient, agents: list[str]) -> Optional[str]:
    presences = client.communications.get_presences_by_user_id(agents).execute_query()
    for e in presences:
        if is_available(e):
            return e.id
    return None


def monitor_presence(client: GraphClient, agents: list[str], cycles: int = 3):
    print(f"Monitoring {len(agents)} agents (poll every {POLL_INTERVAL}s)")
    previous = {}
    for cycle in range(1, cycles + 1):
        presences = client.communications.get_presences_by_user_id(agents).execute_query()
        for e in presences:
            state = agent_status_summary(e)
            old = previous.get(e.id)
            if old is None:
                print(f"  INIT  {e.id}: {state}")
            elif state != old:
                print(f"  CHG   {e.id}: {old} -> {state}")
            previous[e.id] = state
        if cycle < cycles:
            sleep(POLL_INTERVAL)


def main():
    client = GraphClient(tenant=tenant).with_token_interactive(client_id, admin_username)
    agents = [admin_username]

    presences = client.communications.get_presences_by_user_id(agents).execute_query()
    print("Team presence:")
    for e in presences:
        print(f"  {e.id}: {agent_status_summary(e)}")

    available = find_available_agent(client, agents)
    print(f"\nRouting: {'call to ' + available if available else 'no agents — queue'}")

    monitor_presence(client, agents, cycles=3)

    take_agent_offline(client, agents[0])
    print(f"\n{agents[0]} marked offline")


if __name__ == "__main__":
    main()

---

← Back to Microsoft Teams — Communications