# Microsoft Teams API 技能使用 Microsoft Graph API 和 Bot Framework 掌握 Microsoft Teams 自动化。此技能涵盖 Microsoft 365 环境的渠道消息传递、自适应卡、机器人开发、Webhook 和企业集成模式。 ## 何时使用此技能 ### 在以下情况下使用： * 为 Microsoft 365 组织构建机器人 * 创建企业通知系统 * 与Azure DevOps和微软生态系统集成 * 在 Teams 中构建审批工作流程 * 自动化会议安排和管理 * 为 Teams 应用程序创建消息扩展 * 实施合规意识消息解决方案 * 使用自适应卡构建内部工具 ### 不要在以下情况下使用： * 组织主要使用 Slack（使用 slack-api） * 只需要简单的 webhooks（直接使用传入的 webhooks） * 没有可用的 Microsoft 365 订阅 * 构建面向消费者的聊天应用程序 * 需要实时游戏或高频更新 ## 先决条件 ### Azure 应用程序注册 # 1. 转到 Azure 门户 -> Azure Active Directory -> 应用程序注册 # 2.新注册： # - 名称：“Teams Bot 应用程序” # - 支持的帐户类型：此组织目录中的帐户 # - 重定向 URI：Web - https://your-app.azurewebsites.net/auth # 所需的 API 权限 (Microsoft Graph)： # 应用程序权限： # - ChannelMessage.Send - 发送频道消息 # - Chat.ReadWrite.All - 读/写聊天 # - Team.ReadBasic.All - 读取团队信息 # - User.Read.All - 读取用户配置文件 # - Group.Read.All - 读取群组信息 # - OnlineMeetings.ReadWrite.All - 创建会议 # 委托权限： # - Chat.ReadWrite - 用户聊天访问 # - Team.ReadBasic.All - 用户团队访问 # - ChannelMessage.Send - 用户可以发送消息 # 3. 创建客户端密钥： # - 证书和机密 -> 新客户端机密 # - 保存值（仅显示一次） ### Python环境设置 # 创建虚拟环境 python -m venv 团队-bot-env 源 team-bot-env/bin/activate # Linux/macOS # 安装依赖项 pip 安装 azure-identity msgraph-sdk botbuilder-core aiohttp # 创建需求.txt 猫 > 要求.txt << 'EOF' 天蓝色身份>=1.14.0 msgraph-sdk>=1.0.0 botbuilder-core>=4.14.0 botbuilder-integration-aiohttp>=4.14.0 aiohttp>=3.9.0 python-dotenv>=1.0.0 请求>=2.31.0 EOF # 环境变量猫 > .env << 'EOF' AZURE_TENANT_ID=您的租户 ID AZURE_CLIENT_ID=您的客户端 ID AZURE_CLIENT_SECRET=您的客户端秘密 MICROSOFT_APP_ID=您的机器人应用程序 ID MICROSOFT_APP_PASSWORD=您的机器人密码 TEAMS_WEBHOOK_URL=您的 webhook-url EOF ### 机器人框架注册 # 1. 访问 https://dev.botframework.com/bots/new # 2. 或者使用 Azure 门户 -> 创建资源 -> 机器人通道注册 # 机器人配置： # - 消息传递端点：https://your-app.azurewebsites.net/api/messages # - Microsoft 应用程序 ID：来自应用程序注册 # - 启用团队频道 # 使用 ngrok 进行本地开发：恩格洛克 http 3978 # 将消息端点更新为 ngrok URL ## 核心能力 ### 1\. Microsoft Graph API 客户端 # graph_client.py # 关于我：用于 Teams 操作的 Microsoft Graph API 客户端 # ABOUTME：处理身份验证和常见 API 调用从 azure.identity 导入 ClientSecretCredential 从 msgraph 导入 GraphServiceClient 从 msgraph. generated.models.chat_message 导入 ChatMessage 从 msgraph. generated.models.item_body 导入 ItemBody 从 msgraph. generated.models.body_type 导入 BodyType 导入操作系统从 dotenv 导入 load_dotenv 加载_dotenv() TeamsGraphClient 类： """用于 Teams 操作的 Microsoft Graph 客户端""" def __init__(自身): self.credential = ClientSecretCredential( tenant_id=os.environ["AZURE_TENANT_ID"], client_id=os.environ["AZURE_CLIENT_ID"], client_secret=os.environ["AZURE_CLIENT_SECRET"] ） self.client = GraphServiceClient( 凭证=self.credential，范围=[“https://graph.microsoft.com/.default”] ）异步 def send_channel_message( 自我, team_id：str，频道 ID：str，内容：str，内容类型：str =“html” ）： """向 Teams 频道发送消息""" 消息 = 聊天消息( 主体=项目主体（ content_type=BodyType.Html 如果 content_type == "html" else BodyType.Text, 内容=内容））结果 = 等待 self.client.teams.by_team_id(team_id) \ .channels.by_channel_id(channel_id) \ .messages.post(消息) 返回结果异步 def send_chat_message( 自我, 聊天ID：str，内容：str ）： """向聊天室发送消息（1:1 或群组）""" 消息 = 聊天消息( 主体=项目主体（ content_type=BodyType.Html, 内容=内容））结果 = 等待 self.client.chats.by_chat_id(chat_id) \ .messages.post(消息) 返回结果异步 def list_teams(self): """列出应用程序有权访问的所有团队""" 结果 = 等待 self.client.groups.get() team = [g for g in result.value if g.resource_provisioning_options 和 g.resource_provisioning_options 中的“团队”] 返回队异步 def list_channels(self, team_id: str): """列出团队中的频道""" 结果 = 等待 self.client.teams.by_team_id(team_id) \ .channels.get() 返回结果.值异步 def get_channel_messages( 自我, team_id：str，频道 ID：str，顶部：整数 = 50 ）： """从频道获取最近的消息""" 结果 = 等待 self.client.teams.by_team_id(team_id) \ .channels.by_channel_id(channel_id) \ .messages.get( request_configuration=lambda 配置： setattr(config.query_parameters, '顶部', 顶部) ）返回结果.值异步 defreply_to_message( 自我, team_id：str，频道 ID：str，消息 ID：str，内容：str ）： """回复频道消息""" 回复 = 聊天消息( 主体=项目主体（ content_type=BodyType.Html, 内容=内容））结果 = 等待 self.client.teams.by_team_id(team_id) \ .channels.by_channel_id(channel_id) \ .messages.by_chat_message_id(message_id) \ .replies.post(回复) 返回结果异步 def create_online_meeting( 自我, 主题：str，开始时间：str，结束时间：str，参加者：名单）： """创建在线会议""" 从 msgraph. generated.models.online_meeting 导入 OnlineMeeting 从 msgraph. generated.models.meeting_participants 导入 MeetingParticipants 从 msgraph. generated.models.meeting_participant_info 导入 MeetingParticipantInfo 从 msgraph. generated.models.identity_set 导入 IdentitySet 从 msgraph. generated.models.identity 导入身份参与者列表 = [ 会议参加者信息( 身份=身份集（用户=身份（id=与会者）））对于参加者中的参加者 ] 会议 = 在线会议( 主题=主题，开始日期时间=开始时间，结束日期时间=结束时间，参与者=会议参与者( 与会者=participant_list ））结果=等待 self.client.me.online_meetings.post(会议) 返回结果异步 def get_user_by_email(self, email: str): """通过电子邮件获取用户详细信息""" 结果 = 等待 self.client.users.by_user_id(email).get() 返回结果 # 使用示例异步 def main(): 客户端 = TeamsGraphClient() # 列出团队团队 = 等待 client.list_teams() 对于团队中的团队： print(f"团队: {team.display_name} ({team.id})") # 发送频道消息如果团队： team_id = 团队[0].id 频道 = 等待 client.list_channels(team_id) 如果渠道：频道 ID = 频道[0].ID 等待客户端.send_channel_message( 团队 ID，频道 ID, “来自 Python 的问候！这是一条自动消息。” ）如果 __name__ == "__main__": 导入异步 asyncio.run（主（）） ### 2\.自适应卡 # 自适应卡.py # 关于我：用于丰富 Teams 消息的自适应卡片构建 # 关于我：具有操作和数据绑定的互动卡从输入导入字典、列表、可选、任何导入 json 类 AdaptiveCardBuilder： """自适应卡构建器""" def __init__(self, 版本: str = "1.4"): 自我卡 = { “类型”：“自适应卡”， “$schema”：“http://adaptivecards.io/schemas/adaptive-card.json”， “版本”：版本， “身体”：[]， “行动”：[] } def add_text_block( 自我, 文本：str，大小：str =“默认”，权重：str =“默认”，颜色：str =“默认”，换行：布尔 = True ）： """添加文本块""" self.card["body"].append({ “类型”：“文本块”， “文本”：文本， “尺寸”：尺寸， “重量”：重量， “颜色”：颜色， “包裹”：包裹 }) 返回自我 def add_fact_set(self, 事实: Dict[str, str]): """添加事实集（键值对）""" self.card["body"].append({ “类型”：“事实集”， “事实”：[ {“标题”：k，“值”：v} 对于facts.items() 中的 k、v ] }) 返回自我 def add_column_set(self, columns: List[Dict]): """为并排内容添加列集""" self.card["body"].append({ "type": "列集", “列”：列 }) 返回自我 def add_image( 自我, 网址：str，大小：str =“自动”， alt_text：str =“” ）： """添加图片""" self.card["body"].append({ “类型”：“图像”， “网址”：网址， “尺寸”：尺寸， “替代文本”：替代文本 }) 返回自我 def add_action_submit( 自我, 标题：str，数据：字典[str，任意]，样式：str =“默认” ）： """添加提交操作按钮""" self.card["actions"].append({ "type": "操作.提交", “标题”：标题， “数据”：数据， “风格”：风格 }) 返回自我 def add_action_open_url( 自我, 标题：str，网址：str ）： """添加打开 URL 操作""" self.card["actions"].append({ "type": "Action.OpenUrl", “标题”：标题， “网址”：网址 }) 返回自我 def add_action_show_card( 自我, 标题：str，卡：字典）： """添加显示卡片操作（嵌套卡片）""" self.card["actions"].append({ "type": "Action.ShowCard", “标题”：标题， “卡”：卡 }) 返回自我 def add_input_text( 自我, id：str，占位符：str =“”， is_multiline: 布尔 = False, 标签：str =“” ）： """添加文本输入""" 输入元素 = { "type": "输入.文本", “id”： id， “占位符”：占位符， “isMultiline”：is_multiline } 如果标签： input_element["标签"] = 标签 self.card["body"].append(input_element) 返回自我 def add_input_choice_set( 自我, id：str，选择：列表[Dict[str, str]]， is_multi_select: bool = False, 风格：str =“紧凑”，标签：str =“” ）： """Add a choice set (dropdown/radio)""" 输入元素 = { "type": "Input.ChoiceSet", “id”： id， “选择”：选择， "isMultiSelect": is_multi_select, “风格”：风格 } 如果标签： input_element["标签"] = 标签 self.card["body"].append(input_element) 返回自我 def add_container( 自我, 项目：列表[字典]，样式：str =“默认” ）： """添加用于对元素进行分组的容器""" self.card["body"].append({ “类型”：“容器”， “物品”：物品， “风格”：风格 }) 返回自我 def build(self) -> 字典： """Build and return the card""" 归还自助卡 def to_json(self) -> str: """Return card as JSON string""" return json.dumps(self.card, indent=2) def create_deployment_card( 应用程序名称：str，环境：str，版本：str，状态：str，详细信息：字典[str，str]， action_url：str ) -> 字典: """创建部署通知卡""" 状态颜色 = { “成功”：“好”， “失败”：“注意”， "in_progress": "warning", “待处理”：“默认” } 生成器 = AdaptiveCardBuilder() # Header with status color 生成器.add_container([ { “类型”：“文本块”， "text": f"部署 {status.title()}: {app_name}", “尺寸”：“大”， “重量”：“更大胆”， “颜色”：status_colors.get（状态，“默认”） } ], style="emphasis" 如果状态 == "成功" 否则"默认") # 事实 builder.add_fact_set({ “环境”：环境， “版本”：版本， “状态”：status.title(), **详情 }) # 行动 builder.add_action_open_url("查看部署", action_url) 如果状态==“进行中”：生成器.add_action_submit( “取消部署”， {"action": "取消", "app": app_name, "version": 版本}, 风格=“破坏性” ）返回 builder.build() def create_approval_card( 请求 ID：str，标题：str，请求者：str，描述：str，详细信息：字典[str, str] ) -> 字典: """创建批准请求卡""" 生成器 = AdaptiveCardBuilder() 生成器.add_text_block( “需要批准”，尺寸=“大”，重量=“加粗” ） builder.add_text_block（标题，大小=“中”，重量=“粗体”） builder.add_text_block(f"请求者：{requester}") builder.add_text_block（描述，wrap=True）如果详细： builder.add_fact_set（详细信息） # 评论输入生成器.add_input_text( id="评论", placeholder="添加评论（可选）", is_multiline=真，标签=“评论” ） # 批准动作 builder.add_action_submit( “批准”， {"action": "批准", "request_id": request_id}, 风格=“积极” ） builder.add_action_submit( “拒绝”， {"action": "拒绝", "request_id": request_id}, 风格=“破坏性” ） builder.add_action_submit( “请求更多信息”， {“action”：“request_info”，“request_id”：request_id} ）返回 builder.build() def create_poll_card( 问题：str，选项：列表[str]， poll_id：str ) -> 字典: """创建投票卡""" 生成器 = AdaptiveCardBuilder() 生成器.add_text_block( “民意调查”，尺寸=“大”，重量=“加粗” ） builder.add_text_block（问题，size =“medium”，wrap = True）选择=[ {“标题”：选项，“值”：f“选项_{i}”} 对于 i，枚举中的选项（选项） ] builder.add_input_choice_set( id="poll_answer", 选择=选择，样式=“扩展”，标签=“选择你的答案” ）生成器.add_action_submit( “投票”， {“行动”：“投票”，“poll_id”：poll_id} ）返回 builder.build() def create_incident_card( 事件 ID：str，标题：str，严重性：str，描述：str，受影响的服务：列表[str]，时间线：列表[Dict[str, str]] ) -> 字典: """创建事件通知卡""" 严重程度_颜色 = { “批评”：“注意”， “高”：“警告”， “中”：“口音”， “低”：“默认” } 生成器 = AdaptiveCardBuilder() # 标题 builder.add_column_set([ { “类型”：“列”， “宽度”：“自动”， “项目”：[ { “类型”：“文本块”， "text": f"事件：{incident_id}", “尺寸”：“大”， “重量”：“更大胆”， “颜色”：severity_colors.get（严重性，“默认”） } ] }, { “类型”：“列”， “宽度”：“拉伸”， “项目”：[ { “类型”：“文本块”， “文本”：严重程度.upper(), "horizontalAlignment": "右", “重量”：“更大胆”， “颜色”：severity_colors.get（严重性，“默认”） } ] } ]） builder.add_text_block（标题，大小=“中”，重量=“粗体”） builder.add_text_block（描述，wrap=True） # 受影响的服务如果受影响的服务： builder.add_text_block("受影响的服务：",weight="bolder") 对于受影响的服务中的服务： builder.add_text_block(f"- {service}") # 时间轴如果时间线： builder.add_text_block("时间轴：",weight="bolder") 用于进入时间线：生成器.add_text_block( f"**{entry['时间']}**: {entry['更新']}", 换行=真） # 行动生成器.add_action_submit( “承认”， {“操作”：“确认”，“事件ID”：事件ID} ）生成器.add_action_submit( “升级”， {“action”：“升级”，“incident_id”：incident_id}，风格=“破坏性” ）返回 builder.build() ### 3\.传入的 Webhook # webhooks.py # 关于我：团队传入 webhook 集成 # 关于我：简单的通知，无需完整的机器人注册导入请求导入 json 从输入导入字典、列表、可选从日期时间导入日期时间 TeamsWebhook 类： """Microsoft Teams 的传入 Webhook 客户端""" def __init__(self, webhook_url: str): self.webhook_url = webhook_url 定义发送（自我, 文本：str = 无，标题：str = 无，部分：列表[Dict] =无，主题颜色：str = 无，自适应卡：字典=无 ) -> 字典： """通过 webhook 发送消息""" 如果自适应卡： # 发送自适应卡有效负载={ “类型”：“消息”， “附件”：[ { "contentType": "application/vnd.microsoft.card.adaptive", “contentUrl”：无， “内容”：adaptive_card } ] } 其他： # 发送消息卡（传统但更简单）有效负载={ "@type": "留言卡", "@context": "http://schema.org/extensions", “主题颜色”：主题颜色或“0076D7”， “summary”：标题或文本[:50] 如果文本为“通知” } 如果标题：负载[“标题”] = 标题如果文本：有效负载[“文本”] = 文本如果部分：有效负载[“部分”] = 部分响应 = requests.post( self.webhook_url, json=有效负载， headers={"Content-Type": "application/json"} ）如果响应.status_code == 200：返回 {“ok”：True，“status”：response.status_code} 其他：返回{ “好的”：错误， “状态”：响应.status_code， “错误”：响应.文本 } def send_deployment_notification( 自我, 应用程序名称：str，环境：str，版本：str，状态：str， commit_sha：str，作者：str, 部署网址：str = 无，日志网址：str = 无）： """发送部署通知""" 主题颜色={ “成功”：“00FF00”， “失败”：“FF0000”， “开始”：“FFCC00”， “待处理”：“808080” } 状态表情符号= { “成功”：“好的”， “失败”：“X”， “开始”：“火箭”， “待处理”：“时钟” } 部分= [ { "activityTitle": f"部署 {status.title()}: {app_name}", "activitySubtitle": f"作者：{author}", “事实”：[ {“名称”：“环境”，“值”：环境}， {“名称”：“版本”，“值”：版本}， {"名称": "提交", "值": commit_sha[:8]}, {"名称": "时间", "值": datetime.now().strftime("%Y-%m-%d %H:%M:%S")} ], “降价”：正确 } ] 潜在行动 = [] 如果部署网址：潜在的行动.追加（{ "@type": "OpenUri", "name": "查看部署", “目标”：[{“os”：“默认”，“uri”：deploy_url}] }) 如果logs_url：潜在的行动.追加（{ "@type": "OpenUri", "name": "查看日志", “目标”：[{“os”：“默认”，“uri”：logs_url}] }) 如果潜在的行动：部分[0][“潜在行动”] = 潜在行动返回 self.send( title=f"部署{status.title()}", 部分=部分， theme_color=theme_colors.get(状态,“0076D7”) ） def 发送警报( 自我, 标题：str，消息：str，严重性：str =“警告”，来源：str =“系统”，详细信息：可选[Dict] = None， action_url：可选[str] =无）： """发送警报通知""" 严重程度_颜色 = { “关键”：“FF0000”， “错误”：“FF4444”， “警告”：“FFCC00”， “信息”：“0088FF” } 事实=[ {“名称”：“严重性”，“值”：严重性.upper（）}， {“名称”：“来源”，“值”：来源}， {"名称": "时间", "值": datetime.now().strftime("%Y-%m-%d %H:%M:%S")} ] 如果详细：对于键，值在details.items()中： facts.append({"name": key, "value": str(value)}) 部分= [ { “活动标题”：标题， “文本”：消息， “事实”：事实， “降价”：正确 } ] 如果action_url：部分[0][“潜在行动”] = [ { "@type": "OpenUri", "name": "查看详情", “目标”：[{“os”：“默认”，“uri”：action_url}] } ] 返回 self.send( 标题=标题，部分=部分， theme_color=severity_colors.get(严重性, "0076D7") ） def send_adaptive_card(self, 卡: 字典): """发送自适应卡""" 返回 self.send(adaptive_card=card) # 用法如果 __name__ == "__main__": webhook = TeamsWebhook("https://outlook.office.com/webhook/...") # 简单消息 webhook.send(text="来自 Python 的问候！") # 部署通知 webhook.send_deployment_notification（ app_name="我的服务", 环境=“生产”，版本=“v1.2.3”，状态=“成功”， commit_sha=“abc123def456”，作者=“developer@example.com”，部署_url =“https://deployments.example.com” ） # 警报 webhook.send_alert( title="CPU 使用率高", message="服务器 prod-web-01 CPU 使用率超过 90%", 严重性=“警告”，来源=“监控”，详细信息={“服务器”：“prod-web-01”，“当前”：“92%”} ） ### 4\.机器人框架集成 # 机器人.py # 关于我：使用 Bot Framework SDK 的 Teams 机器人 # 关于我：处理消息、卡片和主动消息传递从 botbuilder.core 导入（活动处理程序，转动上下文，卡工厂，消息工厂）从 botbuilder.schema 导入（活动，活动类型，渠道帐户, 附件）导入 json TeamsBot 类（活动处理程序）： """Microsoft Teams 机器人处理程序""" def __init__(self,conversation_references: dict = None): self.conversation_references=conversation_references 或 {} 异步 def on_message_activity(self, Turn_context: TurnContext): """处理传入消息""" # 存储对话参考以主动发送消息 self._add_conversation_reference(turn_context.activity) 文本=turn_context.activity.text.lower().strip() 用户名=turn_context.activity.from_property.name 如果文本==“帮助”：等待 self._send_help_card(turn_context) elif 文本==“状态”：等待 self._send_status_card(turn_context) elif text.startswith(“部署”): 等待 self._handle_deploy_command(turn_context, 文本) 其他：等待turn_context.send_activity（ f“嗨，{user_name}！我收到：‘{text}’。输入‘help’以获取命令。” ）异步 def on_members_add_activity( 自我, members_added：列表， Turn_context：TurnContext ）： """处理添加到对话中的新成员""" 对于members_added中的成员：如果member.id！=turn_context.activity.recipient.id：等待turn_context.send_activity（ f“欢迎加入团队，{member.name}！” “输入‘帮助’来查看可用的命令。” ）异步 def on_adaptive_card_invoke( 自我, Turn_context：TurnContext，调用值：字典）： """处理自适应卡操作调用""" 动作 = invoke_value.get("动作") 数据 = 调用值如果操作==“批准”：返回等待 self._handle_approval（turn_context，数据，批准= True） elif 操作==“拒绝”：返回等待 self._handle_approval（turn_context，数据，批准= False） elif 操作==“投票”：返回等待 self._handle_vote(turn_context, 数据) 返回{“状态”：200} 异步 def _send_help_card(self, Turn_context: TurnContext): """发送帮助卡""" 卡={ “类型”：“自适应卡”， “$schema”：“http://adaptivecards.io/schemas/adaptive-card.json”， “版本”：“1.4”， “身体”：[ { “类型”：“文本块”， "text": "机器人命令", “尺寸”：“大”， “重量”：“更大胆” }, { “类型”：“事实集”， “事实”：[ {"title": "help", "value": "显示此帮助消息"}, {"title": "status", "value": "显示系统状态"}, {"title": "deploy [env]", "value": "触发部署"}, {"title": "民意调查 [问题]", "value": "创建民意调查"} ] } ] } 附件=附件（ content_type="application/vnd.microsoft.card.adaptive", 内容=卡片）等待turn_context.send_activity（ MessageFactory.附件（附件））异步 def _send_status_card(self, Turn_context: TurnContext): """发送状态卡""" 卡={ “类型”：“自适应卡”， “$schema”：“http://adaptivecards.io/schemas/adaptive-card.json”， “版本”：“1.4”， “身体”：[ { “类型”：“文本块”， "text": "系统状态", “尺寸”：“大”， “重量”：“更大胆” }, { "type": "列集", “列”：[ { “类型”：“列”， “宽度”：“拉伸”， “项目”：[ {"type": "TextBlock", "text": "API", "weight": "bolder"}, {"type": "TextBlock", "text": "健康", "color": "好"} ] }, { “类型”：“列”， “宽度”：“拉伸”， “项目”：[ {"type": "TextBlock", "text": "数据库", "weight": "bolder"}, {"type": "TextBlock", "text": "健康", "color": "好"} ] }, { “类型”：“列”， “宽度”：“拉伸”， “项目”：[ {"type": "TextBlock", "text": "缓存", "weight": "bolder"}, {“类型”：“TextBlock”，“文本”：“警告”，“颜色”：“警告”} ] } ] } ], “行动”：[ { "type": "Action.OpenUrl", "title": "查看仪表板", “url”：“https://status.example.com” } ] } 附件=附件（ content_type="application/vnd.microsoft.card.adaptive", 内容=卡片）等待turn_context.send_activity（ MessageFactory.附件（附件））异步 def _handle_deploy_command( 自我, Turn_context：TurnContext，文本：str ）： """处理部署命令""" 部分 = text.split() 环境 = parts[1] if len(parts) > 1 else "staging" 卡={ “类型”：“自适应卡”， “$schema”：“http://adaptivecards.io/schemas/adaptive-card.json”， “版本”：“1.4”， “身体”：[ { “类型”：“文本块”， "text": "确认部署", “尺寸”：“大”， “重量”：“更大胆” }, { “类型”：“文本块”， "text": f"部署到 **{environment}** 环境？", “包裹”：正确 }, { "type": "输入.ChoiceSet", “id”：“版本”， "标签": "版本", “选择”：[ {"title": "v1.2.3 (最新)", "value": "v1.2.3"}, {“标题”：“v1.2.2”，“值”：“v1.2.2”}， {“标题”：“v1.2.1”，“值”：“v1.2.1”} ], “值”：“v1.2.3” } ], “行动”：[ { "type": "操作.提交", "title": "部署", “风格”：“积极”， “数据”：{ “行动”：“部署”，「环境」：环境 } }, { "type": "操作.提交", "title": "取消", “数据”：{“操作”：“取消”} } ] } 附件=附件（ content_type="application/vnd.microsoft.card.adaptive", 内容=卡片）等待turn_context.send_activity（ MessageFactory.附件（附件））异步 def _handle_approval( 自我, Turn_context：TurnContext，数据：字典，批准：布尔）： """处理审批动作""" request_id = data.get("request_id") 评论 = data.get("评论", "") 用户=turn_context.activity.from_property.name status =“已批准”，如果已批准，则“已拒绝” # 更新原卡更新卡 = { “类型”：“自适应卡”， “$schema”：“http://adaptivecards.io/schemas/adaptive-card.json”， “版本”：“1.4”， “身体”：[ { “类型”：“文本块”， "text": f"请求 {status.title()}", “尺寸”：“大”， “重量”：“更大胆”， “颜色”：“好”如果获得批准，否则“注意” }, { “类型”：“文本块”， "text": f"由{用户}", “isSubtle”：正确 } ] } 如果评论： update_card["body"].append({ “类型”：“文本块”， "text": f"评论: {comment}", “包裹”：正确 }) # 返回更新后的卡片返回{ “状态代码”：200， “类型”：“应用程序/vnd.microsoft.card.adaptive”， “值”：updated_card } 异步def _handle_vote（自我，turn_context：TurnContext，数据：dict）： """处理投票""" poll_id = data.get("poll_id") 答案 = data.get("poll_answer") 用户=turn_context.activity.from_property.name 等待turn_context.send_activity（ f“{user} 投票：{answer}” ）返回{“状态代码”：200} def _add_conversation_reference（自身，活动：活动）： """存储对话参考以主动发送消息""" conversation_reference = TurnContext.get_conversation_reference(活动) self.conversation_references[ 对话参考.对话.id ] = 对话参考 ### 5\.主动消息传递 # 主动.py # 关于我：向 Teams 渠道和用户发送主动消息 # 关于我：通知用户而不需要他们发起对话从 botbuilder.core 导入 TurnContext 从 botbuilder.core.teams 导入 TeamsInfo 从 botbuilder.schema 导入活动、ConversationReference 从 botbuilder.integration.aiohttp 导入 CloudAdapter、ConfigurationBotFrameworkAuthentication 导入异步从输入导入字典 ProactiveMessenger 类： """向团队发送主动消息""" def __init__( 自我, 适配器：云适配器，应用程序 ID：str， conversation_references: Dict[str, ConversationReference] ）： self.adapter = 适配器 self.app_id = app_id self.conversation_references = 对话引用异步 def send_to_conversation( 自我, 对话 ID：str，消息：str = 无，卡：字典=无）： """向存储的对话发送主动消息""" 如果 conversation_id 不在 self.conversation_references 中： raise ValueError(f"没有 {conversation_id} 的对话参考") 对话参考 = self.对话参考[对话 ID] 异步 def 回调(turn_context: TurnContext): 如果卡：从 botbuilder.schema 导入附件从 botbuilder.core 导入 MessageFactory 附件=附件（ content_type="application/vnd.microsoft.card.adaptive", 内容=卡片）等待turn_context.send_activity（ MessageFactory.附件（附件））其他：等待turn_context.send_activity（消息）等待 self.adapter. continue_conversation( 对话参考，回调， self.app_id ）异步 def send_to_channel( 自我, 服务网址：str， team_id：str，频道 ID：str，消息：str = 无，卡：字典=无）： """向 Teams 频道发送主动消息""" 从 botbuilder.schema 导入（对话参数，活动，渠道账户） # 为频道创建对话参考对话参数 = 对话参数( is_group=真， channel_data={"channel": {"id":channel_id}}, 活动=活动（类型=“消息”，文本=消息 ) 如果消息 else 无）异步 def 回调(turn_context: TurnContext): 如果卡：从 botbuilder.schema 导入附件从 botbuilder.core 导入 MessageFactory 附件=附件（ content_type="application/vnd.microsoft.card.adaptive", 内容=卡片）等待turn_context.send_activity（ MessageFactory.附件（附件））埃利夫消息：等待turn_context.send_activity（消息） # 创建并发送主动消息对话参考 = 对话参考( service_url=服务_url, channel_id="msteams", 对话={“id”：channel_id} ）等待 self.adapter. continue_conversation( 对话参考，回调， self.app_id ）异步 def notify_all_conversations( 自我, 消息：str = 无，卡：字典=无）： """向所有存储的对话广播消息""" 对于 self.conversation_references 中的 conv_id：尝试：等待 self.send_to_conversation(conv_id, 消息, 卡) 除了异常 e： print(f"通知 {conv_id} 失败：{e}") # Azure Functions 的使用示例 ”“” # 函数_app.py 导入 azure.functions 作为 func 从主动导入 ProactiveMessenger async def notification_deployment_complete(req: func.HttpRequest) -> func.HttpResponse: # 加载配置和适配器信使= ProactiveMessenger（适配器，app_id，conversation_references）卡 = 创建部署卡（ app_name="我的服务", 环境=“生产”，版本=“v1.2.3”，状态=“成功” ）等待信使.send_to_channel( service_url =“https://smba.trafficmanager.net/teams/”， team_id="您的团队 ID", Channel_id="您的频道 ID", 卡=卡） return func.HttpResponse("通知已发送", status_code=200) ”“” ### 6\.会议自动化 # 会议.py # 关于我：团队通过 Graph API 实现自动化会议 # 关于我：创建、管理和获取会议详细信息从日期时间导入日期时间，时间增量从输入导入列表、可选、字典导入异步类会议管理器： """通过 Graph API 管理 Teams 会议""" def __init__(self, graph_client): self.client = graph_client 异步 def create_instant_meeting( 自我, 主题：str，组织者 ID：str ) -> 字典: """创建即时会议""" 从 msgraph. generated.models.online_meeting 导入 OnlineMeeting 会议 = 在线会议( 主题=主题， start_date_time=datetime.utcnow().isoformat() + "Z", end_date_time=(datetime.utcnow() + timedelta(小时=1)).isoformat() + "Z" ）结果 = 等待 self.client.users.by_user_id(organizer_id) \ .online_meetings.post（会议）返回{ “join_url”：结果.join_web_url， “会议id”：结果.id， “主题”：结果.主题 } 异步 def Schedule_meeting( 自我, 主题：str，开始时间：日期时间，结束时间：日期时间，组织者 ID：str， attendee_emails：列表[str]，正文：str =“” ) -> 字典: """安排与与会者的会议""" 从 msgraph. generated.models.event 导入事件从 msgraph. generated.models.item_body 导入 ItemBody 从 msgraph. generated.models.body_type 导入 BodyType 从 msgraph. generated.models.attendee 导入 attendee 从 msgraph. generated.models.email_address 导入 EmailAddress 从 msgraph. generated.models.attendee_type 导入 attendeeType 从 msgraph. generated.models.date_time_time_zone 导入 DateTimeTimeZone 参加者 = [ 出席者( email_address=电子邮件地址(地址=电子邮件), 类型=AttendeeType.必填）对于 attendee_emails 中的电子邮件 ] 事件 = 事件（主题=主题，主体=项目主体（ content_type=BodyType.Html, 内容=正文 ), 开始=日期时间时区( date_time=start_time.isoformat(), 时区=“世界标准时间” ), 结束=日期时间时区( date_time=end_time.isoformat(), 时区=“世界标准时间” ), 参加者=参加者， is_online_meeting=正确, online_meeting_provider="teamsForBusiness" ）结果 = 等待 self.client.users.by_user_id(organizer_id) \ .events.post(事件) 返回{ “event_id”：结果.id， “主题”：结果.主题， "join_url": result.online_meeting.join_url if result.online_meeting else None } 异步 def get_user_calendar( 自我, 用户 ID：str，开始日期：日期时间，结束日期：日期时间 ) -> 列表[字典]: """获取用户的日历事件""" 结果 = 等待 self.client.users.by_user_id(user_id) \ .calendar_view.get( request_configuration=lambda 配置：（ setattr(config.query_parameters, 'start_date_time', start_date.isoformat()), setattr(config.query_parameters, 'end_date_time', end_date.isoformat()) ））返回[ { "id": 事件.id, “主题”：事件.主题， “开始”：event.start.date_time， “结束”：event.end.date_time， “is_online”：event.is_online_meeting } 对于 result.value 中的事件 ] 异步 def cancel_meeting( 自我, 用户 ID：str，事件 ID：str，取消消息：str =“” ）： """取消预定的会议""" 等待 self.client.users.by_user_id(user_id) \ .events.by_event_id(event_id) \ .cancel.post(comment=cancellation_message) # 使用示例异步 def Schedule_standup(): """安排每日站立会议""" 从 graph_client 导入 TeamsGraphClient 客户端 = TeamsGraphClient() 会议 = MeetingManager(client.client) # 明天上午 9 点的时间表明天 = datetime.utcnow().replace(小时=9, 分钟=0) + timedelta(天=1) 结果=等待会议.schedule_meeting( 主题=“每日站立”，开始时间=明天， end_time=明天 + timedelta(分钟=30), organizer_id="organizer@company.com", 与会者电子邮件=[ “team-member1@company.com”， “team-member2@company.com” ], body="

每日站会

请准备好分享您的最新动态。

" ） print(f"已安排的会议：{result['join_url']}") ## 集成示例 ### Azure DevOps 管道集成 # azure-pipelines.yml 触发： - 主要池： vmImage: 'ubuntu-最新' 阶段： - 阶段：构建职位： - 工作：BuildJob 步骤： - 脚本：回显“正在构建...” - 任务：PowerShell@2 displayName: '通知团队 - 构建已开始' 输入：目标类型：'内联' 脚本： | $webhook = "$(TEAMS_WEBHOOK_URL)" $正文=@{ “@type”=“留言卡” “@context”=“http://schema.org/extensions” “主题颜色”=“FFCC00” “摘要”=“构建已开始” “部分”= @( @{ "activityTitle" = "构建已开始：$(Build.DefinitionName)" “事实”=@( @{“名称”=“分行”; “值” = “$(Build.SourceBranchName)” } @{“名称”=“提交”; “值” = “$(Build.SourceVersion)” } @{“名称”=“构建 ID”; “值” = “$(Build.BuildId)” } ） } ） } |转换为 Json -深度 10 Invoke-RestMethod -Uri $webhook -Method Post -Body $body -ContentType 'application/json' - 阶段：部署取决于：构建职位： - 部署：DeployJob 环境：“生产” 策略：运行一次：部署：步骤： - 脚本：echo“正在部署...” - 任务：PowerShell@2 displayName: '通知团队 - 部署完成' 输入：目标类型：'内联' 脚本： | $webhook = "$(TEAMS_WEBHOOK_URL)" $正文=@{ “@type”=“留言卡” “主题颜色”=“00FF00” “摘要”=“部署完成” “部分”= @( @{ “activityTitle”=“部署完成” “事实”=@( @{“名称”=“环境”; “价值”=“产量”} @{“名称”=“版本”; “值” = “$(Build.BuildNumber)” } ） “潜在行动”= @( @{ “@type”=“OpenUri” “名称”=“查看版本” “目标” = @(@{ “os” = “默认”; “uri” = “$(System.TeamFoundationCollectionUri)/$(System.TeamProject)/_release?releaseId=$(Release.ReleaseId)” }) } ） } ） } |转换为 Json -深度 10 Invoke-RestMethod -Uri $webhook -Method Post -Body $body -ContentType 'application/json' ### FastAPI 机器人端点 # main.py # 关于我：Teams 机器人的 FastAPI 端点 # 关于我：处理机器人消息和卡片操作从 fastapi 导入 FastAPI、请求、响应从 botbuilder.core 导入 TurnContext 从 botbuilder.integration.aiohttp 导入 CloudAdapter、ConfigurationBotFrameworkAuthentication 从 botbuilder.schema 导入活动从机器人导入 TeamsBot 导入操作系统 # 配置类默认配置：端口 = 3978 APP_ID = os.environ.get("MICROSOFT_APP_ID", "") APP_PASSWORD = os.environ.get("MICROSOFT_APP_PASSWORD", "") 配置=默认配置（） # 创建适配器设置 = ConfigurationBotFrameworkAuthentication(CONFIG) 适配器 = CloudAdapter（设置） # 创建机器人 CONVERSATION_REFERENCES = {} BOT = TeamsBot(CONVERSATION_REFERENCES) # 错误处理程序 async def on_error(上下文: TurnContext, 错误: 异常): print(f"机器人错误：{错误}") wait context.send_activity("抱歉，发生错误。") ADAPTER.on_turn_error = on_error # FastAPI 应用程序应用程序 = FastAPI() @app.post(“/api/messages”) 异步 def 消息（请求：请求）-> 响应： """主要机器人消息端点""" 如果“application/json”不在 request.headers.get("Content-Type", "") 中：返回响应（status_code=415） body = 等待 request.json() 活动 = Activity().反序列化(body) auth_header = request.headers.get("授权", "") 响应 = 等待 ADAPTER.process_activity(auth_header, 活动, BOT.on_turn) 如果回应：返回响应（内容=响应.body, status_code=响应.status ）返回响应（status_code=201） @app.get(“/api/health”) 异步 def health(): 返回{“状态”：“健康”} 如果 __name__ == "__main__": 进口uvicorn uvicorn.run（应用程序，主机=“0.0.0.0”，端口= CONFIG.PORT） ## 最佳实践 ### 1\.速率限制 # 尊重 Graph API 速率限制导入时间从 functools 导入包装 defrate_limit_handler(max_retries=3): def 装饰器（函数）： @wraps(函数) 异步 def 包装器（*args，**kwargs）：对于范围内的尝试（max_retries）：尝试：返回等待函数（*args，**kwargs）除了异常 e： if "429" in str(e): # 请求过多延迟 = 2 ** 尝试等待 asyncio.sleep（延迟）其他：提高引发异常（“超出最大重试次数”）返回包装器返回装饰器 ### 2\.代币管理 # 安全令牌处理从 azure.identity 导入 DefaultAzureCredential 从 functools 导入 lru_cache @lru_cache() def get_credential(): """获取缓存的 Azure 凭据""" 返回默认AzureCredential() # 在生产中使用托管身份 # 为本地开发使用环境变量 ### 3\.卡片设计 # 自适应卡最佳实践 def create_accessible_card(): 返回{ “类型”：“自适应卡”， “版本”：“1.4”， “身体”：[ { “类型”：“文本块”， "text": "重要信息", “尺寸”：“大”， “重量”：“更大胆”， # 始终包含后备文本 “后备”：“下降” } ], # 为老客户端提供后备 "fallbackText": "此卡需要较新的 Teams 客户端。" } ## 故障排除 ### 常见问题 **问题：机器人未接收消息** # 验证消息端点是否可访问卷曲 -X POST https://your-bot.azurewebsites.net/api/messages \ -H“内容类型：application/json”\ -d '{"type": "ping"}' # 检查Azure应用程序注册权限 # 验证 Bot Framework 中是否启用了 Teams 通道 **问题：Webhook 返回错误** # 调试 webhook 响应响应 = requests.post(webhook_url, json=payload) print(f"状态：{response.status_code}") print(f"响应：{response.text}") **问题：Graph API 权限被拒绝** # 验证API权限是否被授予 # 应用程序权限可能需要管理员同意 # 检查令牌范围是否匹配所需的权限 ## 版本历史版本 |日期 |变化 ---|---|--- 1.0.0 | 2026-01-17 |具有全面 Teams API 模式的初始版本 ## 资源 * [Microsoft Graph API]() * [机器人框架 SDK]() * [自适应卡片设计器]() * [Teams 应用清单]() * [Teams Webhook 连接器]() ** ** _此技能为 Microsoft Teams 自动化提供生产就绪模式，从而实现企业消息传递和协作工作流程。_

团队API

包含技能