如何将MS Teams创建的通知接收至Python Django Web应用并存储至数据库

当然可以实现，下面分两种常用方案说明具体步骤：

方案一：使用Teams Outgoing Webhook（适合团队内特定消息触发） #

这种方式适用于监听某个Teams团队/频道内的消息，当有新消息时自动推送到你的Django应用。

1. 在Teams中创建Outgoing Webhook #

• 打开目标Teams团队，进入「管理团队」>「应用」>「创建应用」>「Outgoing Webhook」
• 填写名称、描述，上传图标，输入你的Django应用公网可访问的API端点URL（例如https://your-domain.com/api/teams-webhook/）
• 创建完成后，保存系统生成的安全令牌（用于后续请求合法性验证）

2. Django端实现接收与存储 #

第一步：创建数据模型

# models.py
from django.db import models

class TeamsNotification(models.Model):
    message_id = models.CharField(max_length=255, unique=True)
    sender_name = models.CharField(max_length=255)
    sender_email = models.EmailField(null=True, blank=True)
    content = models.TextField()
    channel_name = models.CharField(max_length=255)
    created_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return f"{self.sender_name}: {self.content[:50]}"

执行迁移命令创建表：

python manage.py makemigrations
python manage.py migrate

第二步：编写接收视图

# views.py
import hmac
import hashlib
from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST
from .models import TeamsNotification

# 替换为你在Teams中获取的Outgoing Webhook令牌
TEAMS_WEBHOOK_SECRET = "your-teams-webhook-secret"

@csrf_exempt
@require_POST
def teams_webhook_receiver(request):
    # 验证请求签名，防止伪造
    auth_header = request.headers.get("Authorization", "")
    signature = auth_header.split(" ")[-1] if auth_header else ""
    
    computed_signature = hmac.new(
        TEAMS_WEBHOOK_SECRET.encode(),
        request.body,
        hashlib.sha256
    ).hexdigest()
    
    if not hmac.compare_digest(signature, computed_signature):
        return JsonResponse({"error": "Invalid request signature"}, status=403)
    
    # 解析Teams发送的JSON数据
    data = request.json()
    message_id = data["id"]
    sender_name = data["from"]["name"]
    sender_email = data["from"].get("user", {}).get("email")
    content = data["text"]
    channel_name = data["conversation"]["name"]
    
    # 存储到数据库
    TeamsNotification.objects.create(
        message_id=message_id,
        sender_name=sender_name,
        sender_email=sender_email,
        content=content,
        channel_name=channel_name
    )
    
    return JsonResponse({"status": "success"}, status=200)

第三步：配置URL路由

# urls.py
from django.urls import path
from .views import teams_webhook_receiver

urlpatterns = [
    path("api/teams-webhook/", teams_webhook_receiver, name="teams-webhook"),
]

方案二：使用Microsoft Graph API订阅（监听全量Teams事件） #

如果需要监听更广泛的Teams事件（如团队成员变更、频道创建、所有频道消息等），可以用Graph API的订阅功能。

1. 注册Azure AD应用 #

• 登录Azure门户，创建新的应用注册，获取客户端ID、客户端密钥和租户ID
• 配置应用API权限：添加Microsoft Graph的ChannelMessage.Read.All（或对应你需要的事件权限），并授予管理员同意

2. Django端实现订阅与接收 #

第一步：获取Graph API访问令牌

# utils.py
import requests

def get_graph_access_token():
    client_id = "your-azure-ad-client-id"
    client_secret = "your-azure-ad-client-secret"
    tenant_id = "your-tenant-id"
    
    token_url = f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/token"
    payload = {
        "grant_type": "client_credentials",
        "client_id": client_id,
        "client_secret": client_secret,
        "scope": "https://graph.microsoft.com/.default"
    }
    
    response = requests.post(token_url, data=payload)
    response.raise_for_status()
    return response.json()["access_token"]

第二步：创建Graph订阅

# utils.py 继续添加
def create_teams_message_subscription(team_id, channel_id):
    token = get_graph_access_token()
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }
    
    # 订阅指定频道的新消息事件
    payload = {
        "changeType": "created",
        "notificationUrl": "https://your-domain.com/api/graph-webhook/",
        "resource": f"/teams/{team_id}/channels/{channel_id}/messages",
        "expirationDateTime": "2024-12-31T18:00:00Z",
        "clientState": "your-random-validation-string"
    }
    
    response = requests.post("https://graph.microsoft.com/v1.0/subscriptions", headers=headers, json=payload)
    response.raise_for_status()
    return response.json()

第三步：编写回调视图处理通知

# views.py
from django.http import HttpResponse, JsonResponse
from django.views.decorators.csrf import csrf_exempt
from .models import TeamsNotification
from .utils import get_graph_access_token
import requests

@csrf_exempt
def graph_webhook_receiver(request):
    # 处理订阅验证请求
    if request.method == "GET":
        validation_token = request.GET.get("validationToken")
        if validation_token:
            return HttpResponse(validation_token, content_type="text/plain")
        return HttpResponse(status=400)
    
    # 处理事件通知
    if request.method == "POST":
        notifications = request.json()["value"]
        token = get_graph_access_token()
        
        for notification in notifications:
            # 获取完整的消息详情
            resource_url = notification["resource"]
            headers = {"Authorization": f"Bearer {token}"}
            message_response = requests.get(f"https://graph.microsoft.com/v1.0{resource_url}", headers=headers)
            message_response.raise_for_status()
            message_data = message_response.json()
            
            # 存储到数据库
            TeamsNotification.objects.create(
                message_id=message_data["id"],
                sender_name=message_data["from"]["user"]["displayName"],
                sender_email=message_data["from"]["user"]["email"],
                content=message_data["body"]["content"],
                channel_name=message_data["channelIdentity"]["displayName"],
                created_at=message_data["createdDateTime"]
            )
        
        return JsonResponse({"status": "success"}, status=202)

第四步：配置路由

# urls.py
from django.urls import path
from .views import teams_webhook_receiver, graph_webhook_receiver

urlpatterns = [
    path("api/teams-webhook/", teams_webhook_receiver, name="teams-webhook"),
    path("api/graph-webhook/", graph_webhook_receiver, name="graph-webhook"),
]

关键注意事项 #

• 你的Django应用必须有公网可访问的URL，本地开发可使用ngrok等工具映射本地端口到公网
• 务必验证请求合法性，避免恶意请求攻击
• Graph API订阅有过期时间，需定期调用续订接口延长有效期
• 根据需求调整Graph API的权限和订阅的资源路径

内容的提问来源于stack exchange，提问作者Jan Kens