使用 Microsoft Graph 获取 Teams 未读聊天消息
结论
Microsoft Graph 可以返回当前登录用户在某个聊天中的最后阅读时间。这个时间记录在 chat 的 viewpoint.lastMessageReadDateTime 属性中，单条 chatMessage 和聊天成员对象都不包含该信息。
获取这个属性需要满足两个条件：
- 使用
Chat.Read等委托权限，以用户身份获取令牌。 - 查询聊天时显式请求
viewpoint。
Microsoft Graph 不直接返回聊天的“未读消息数量”。要计算未读消息，通常需要比较 viewpoint.lastMessageReadDateTime 和消息的 createdDateTime。
使用客户端凭据等 application-only 身份时，Graph 无法识别“当前用户”的阅读位置，也就无法通过 viewpoint 获取个人未读状态。即使使用管理员身份或取得管理员同意，这一限制仍然存在。
为什么当前代码取不到阅读时间
lastReadDateTime 和 lastMessageReadDateTime 都不属于 chatMessage，因此下面的写法取不到结果：
m.get("lastReadDateTime")/chats/{chat-id}/members 返回聊天成员信息，其中也没有每个成员的阅读回执。阅读时间位于：
chat.viewpoint.lastMessageReadDateTime
查询聊天时需要显式选择该属性：
GET https://graph.microsoft.com/v1.0/me/chats?$select=id,topic,chatType,viewpoint
Authorization: Bearer <token>响应结构通常如下：
{
"id": "19:...",
"chatType": "oneOnOne",
"topic": null,
"viewpoint": {
"isHidden": false,
"lastMessageReadDateTime": "2026-10-28T08:25:41.123Z"
}
}如果 viewpoint 仍然为空，先确认访问令牌是否由委托登录流程签发，而非 application-only 令牌。
使用 MSAL 持久化登录状态
每次调用 acquire_token_interactive() 都会打开登录窗口。可以持久化保存 MSAL 令牌缓存，并在程序启动时先调用 acquire_token_silent()。访问令牌过期后，MSAL 会利用缓存中的账户信息和刷新令牌进行续期。
下面的示例会持久化 MSAL 缓存并静默获取令牌，然后查询 viewpoint.lastMessageReadDateTime，筛选创建时间晚于最后阅读时间的消息，同时排除自己发送的消息。
import json
import os
import time
from datetime import datetime
from pathlib import Path
import msal
import requests
CLIENT_ID = "YOUR_CLIENT_ID"
TENANT_ID = "YOUR_TENANT_ID"
AUTHORITY = f"https://login.microsoftonline.com/{TENANT_ID}"
SCOPES = ["Chat.Read"]
GRAPH_ROOT = "https://graph.microsoft.com/v1.0"
POLL_INTERVAL = 30
CACHE_FILE = Path.home() / ".cache" / "teams-unread" / "msal-cache.json"
token_cache = msal.SerializableTokenCache()
if CACHE_FILE.exists():
token_cache.deserialize(CACHE_FILE.read_text(encoding="utf-8"))
app = msal.PublicClientApplication(
CLIENT_ID,
authority=AUTHORITY,
token_cache=token_cache,
)
def save_token_cache():
if not token_cache.has_state_changed:
return
CACHE_FILE.parent.mkdir(parents=True, exist_ok=True)
CACHE_FILE.write_text(token_cache.serialize(), encoding="utf-8")
os.chmod(CACHE_FILE, 0o600)
def get_access_token():
accounts = app.get_accounts()
result = None
if accounts:
result = app.acquire_token_silent(
scopes=SCOPES,
account=accounts[0],
)
if not result:
result = app.acquire_token_interactive(scopes=SCOPES)
save_token_cache()
if "access_token" not in result:
raise RuntimeError(
result.get("error_description", "Microsoft 登录失败")
)
return result["access_token"]
def graph_get(url, params=None):
while True:
headers = {
"Authorization": f"Bearer {get_access_token()}",
"Accept": "application/json",
}
response = requests.get(
url,
headers=headers,
params=params,
timeout=30,
)
if response.status_code == 429:
retry_after = int(response.headers.get("Retry-After", "30"))
time.sleep(retry_after)
continue
response.raise_for_status()
return response.json()
def get_collection(url, params=None):
items = []
while url:
data = graph_get(url, params=params)
items.extend(data.get("value", []))
url = data.get("@odata.nextLink")
params = None
return items
def parse_graph_time(value):
return datetime.fromisoformat(value.replace("Z", "+00:00"))
def list_chats():
return get_collection(
f"{GRAPH_ROOT}/me/chats",
params={
"$select": "id,topic,chatType,viewpoint",
},
)
def get_current_user_id():
data = graph_get(
f"{GRAPH_ROOT}/me",
params={"$select": "id"},
)
return data["id"]
def get_unread_messages(chat, current_user_id):
viewpoint = chat.get("viewpoint") or {}
last_read_value = viewpoint.get("lastMessageReadDateTime")
if not last_read_value:
# Graph 没有返回可用的阅读位置时，不应把全部历史记录
# 直接判定为未读。
return []
last_read = parse_graph_time(last_read_value)
messages = get_collection(
f"{GRAPH_ROOT}/chats/{chat['id']}/messages",
params={
"$orderby": "createdDateTime desc",
"$top": "50",
},
)
unread = []
for message in messages:
created_value = message.get("createdDateTime")
if not created_value:
continue
created = parse_graph_time(created_value)
if created <= last_read:
continue
sender = (message.get("from") or {}).get("user") or {}
if sender.get("id") == current_user_id:
continue
# 排除成员变更等系统事件
if message.get("messageType") != "message":
continue
unread.append(message)
return unread
def main():
current_user_id = get_current_user_id()
while True:
total_unread = 0
for chat in list_chats():
unread = get_unread_messages(chat, current_user_id)
if not unread:
continue
total_unread += len(unread)
name = chat.get("topic") or chat.get("chatType") or chat["id"]
print(f"{name}: {len(unread)} 条未读消息")
for message in sorted(
unread,
key=lambda item: item["createdDateTime"],
):
sender = (
((message.get("from") or {}).get("user") or {})
.get("displayName", "未知发送者")
)
content = (message.get("body") or {}).get("content", "")
print(f" {sender}: {content}")
if total_unread == 0:
print("当前没有检测到未读聊天消息")
time.sleep(POLL_INTERVAL)
if __name__ == "__main__":
try:
main()
finally:
save_token_cache()应用注册需要启用公共客户端登录，并配置 Microsoft Graph 委托权限 Chat.Read。遇到组织的条件访问策略、多重身份验证、密码变更或管理员撤销会话时，用户仍可能需要重新登录。令牌缓存可以避免程序正常启动时反复登录，但不能绕过这些安全策略。
如果运行环境不方便使用浏览器交互，可以将交互式登录改为设备代码流程。首次运行仍需用户完成授权，后续运行则可以继续使用同一份持久化缓存。
关于“未读”的准确性
计算方式为：
消息 createdDateTime > viewpoint.lastMessageReadDateTime
实际使用时还要处理以下边界情况：
- 自己发送的消息通常不计入未读提醒。
- 系统事件、删除消息和应用消息可能没有普通用户发送者。
body.content通常是 HTML，直接输出可能带有标签。- 未读消息数量超过
$top时，需要继续读取@odata.nextLink。示例中的get_collection()已经处理了分页。 lastMessageReadDateTime取决于 Teams 的阅读状态，其更新可能存在延迟。- 该值为空时，不宜把整个聊天历史全部视为未读。首次运行可以建立本地基线，此后只检测新增消息。
Graph 提供的是最后阅读时间，没有权威的 unreadCount 字段，所以这里得到的未读数量是根据时间戳推算的。
降低轮询带来的问题
每隔 10 秒遍历全部聊天和消息会增加请求量，也可能触发 Graph 限流。用于个人桌面提醒时，可以将轮询间隔设为 30 至 60 秒，并按照 429 Too Many Requests 响应中的 Retry-After 等待后重试。
需要长期运行并接近实时接收消息时，可以通过 Microsoft Graph change notifications 订阅聊天消息变化。这种方案需要提供 Graph 能够访问的 HTTPS 回调地址，并定期续订。通知只能表明消息发生了变化；要判断消息是否已读，仍需查询 viewpoint 或维护本地状态。
如果只是要实现 Linux 托盘提醒，也可以维护本地消息水位，记录上次看到的消息 ID 或时间。Graph 返回更新的消息后，再调用 notify-send 或托盘程序发出提示。这种方式可以可靠发现“脚本启动后出现的新消息”，却无法知道用户是否已在 Teams、Firefox 或其他设备上读过，因此不能替代 Teams 的真实阅读状态。
备注：内容仅供参考。