返回 JoyAI-Echo
msteams.py
1 """Microsoft Teams channel MVP using a tiny built-in HTTP webhook server.
2
3 Scope:
4 - DM-focused MVP
5 - text inbound/outbound
6 - conversation reference persistence
7 - sender allowlist support
8 - optional inbound Bot Framework bearer-token validation
9 - no attachments/cards/polls yet
10 """
11
12 from __future__ import annotations
13
14 import asyncio
15 import html
16 import importlib.util
17 import json
18 import re
19 import threading
20 import time
21 from dataclasses import dataclass
22 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
23 from typing import TYPE_CHECKING, Any
24
25 import httpx
26 from loguru import logger
27 from pydantic import Field
28
29 from nanobot.bus.events import OutboundMessage
30 from nanobot.bus.queue import MessageBus
31 from nanobot.channels.base import BaseChannel
32 from nanobot.config.paths import get_workspace_path
33 from nanobot.config.schema import Base
34
35 MSTEAMS_AVAILABLE = (
36 importlib.util.find_spec("jwt") is not None
37 and importlib.util.find_spec("cryptography") is not None
38 )
39
40 if TYPE_CHECKING:
41 import jwt
42
43 if MSTEAMS_AVAILABLE:
44 import jwt
45
46
47 class MSTeamsConfig(Base):
48 """Microsoft Teams channel configuration."""
49
50 enabled: bool = False
51 app_id: str = ""
52 app_password: str = ""
53 tenant_id: str = ""
54 host: str = "0.0.0.0"
55 port: int = 3978
56 path: str = "/api/messages"
57 allow_from: list[str] = Field(default_factory=list)
58 reply_in_thread: bool = True
59 mention_only_response: str = "Hi — what can I help with?"
60 validate_inbound_auth: bool = True
61
62
63 @dataclass
64 class ConversationRef:
65 """Minimal stored conversation reference for replies."""
66
67 service_url: str
68 conversation_id: str
69 bot_id: str | None = None
70 activity_id: str | None = None
71 conversation_type: str | None = None
72 tenant_id: str | None = None
73
74
75 class MSTeamsChannel(BaseChannel):
76 """Microsoft Teams channel (DM-first MVP)."""
77
78 name = "msteams"
79 display_name = "Microsoft Teams"
80
81 @classmethod
82 def default_config(cls) -> dict[str, Any]:
83 return MSTeamsConfig().model_dump(by_alias=True)
84
85 def __init__(self, config: Any, bus: MessageBus):
86 if isinstance(config, dict):
87 config = MSTeamsConfig.model_validate(config)
88 super().__init__(config, bus)
89 self.config: MSTeamsConfig = config
90 self._loop: asyncio.AbstractEventLoop | None = None
91 self._server: ThreadingHTTPServer | None = None
92 self._server_thread: threading.Thread | None = None
93 self._http: httpx.AsyncClient | None = None
94 self._token: str | None = None
95 self._token_expires_at: float = 0.0
96 self._botframework_openid_config_url = (
97 "https://login.botframework.com/v1/.well-known/openidconfiguration"
98 )
99 self._botframework_openid_config: dict[str, Any] | None = None
100 self._botframework_openid_config_expires_at: float = 0.0
101 self._botframework_jwks: dict[str, Any] | None = None
102 self._botframework_jwks_expires_at: float = 0.0
103 self._refs_path = get_workspace_path() / "state" / "msteams_conversations.json"
104 self._refs_path.parent.mkdir(parents=True, exist_ok=True)
105 self._conversation_refs: dict[str, ConversationRef] = self._load_refs()
106
107 async def start(self) -> None:
108 """Start the Teams webhook listener."""
109 if not MSTEAMS_AVAILABLE:
110 logger.error("PyJWT not installed. Run: pip install echo-director-agent[msteams]")
111 return
112
113 if not self.config.app_id or not self.config.app_password:
114 logger.error("MSTeams app_id/app_password not configured")
115 return
116
117 if not self.config.validate_inbound_auth:
118 logger.warning(
119 "MSTeams inbound auth validation was explicitly DISABLED in config. "
120 "Anyone who knows the webhook URL can send messages as any user. "
121 "Only disable this for local development or controlled testing."
122 )
123
124 self._loop = asyncio.get_running_loop()
125 self._http = httpx.AsyncClient(timeout=30.0)
126 self._running = True
127
128 channel = self
129
130 class Handler(BaseHTTPRequestHandler):
131 def do_POST(self) -> None:
132 if self.path != channel.config.path:
133 self.send_response(404)
134 self.end_headers()
135 return
136
137 try:
138 length = int(self.headers.get("Content-Length", "0"))
139 raw = self.rfile.read(length) if length > 0 else b"{}"
140 payload = json.loads(raw.decode("utf-8"))
141 except Exception as e:
142 logger.warning("MSTeams invalid request body: {}", e)
143 self.send_response(400)
144 self.end_headers()
145 return
146
147 auth_header = self.headers.get("Authorization", "")
148 if channel.config.validate_inbound_auth:
149 try:
150 fut = asyncio.run_coroutine_threadsafe(
151 channel._validate_inbound_auth(auth_header, payload),
152 channel._loop,
153 )
154 fut.result(timeout=15)
155 except Exception as e:
156 logger.warning("MSTeams inbound auth validation failed: {}", e)
157 self.send_response(401)
158 self.send_header("Content-Type", "application/json")
159 self.end_headers()
160 self.wfile.write(b'{"error":"unauthorized"}')
161 return
162 try:
163 fut = asyncio.run_coroutine_threadsafe(
164 channel._handle_activity(payload),
165 channel._loop,
166 )
167 fut.result(timeout=15)
168 except Exception as e:
169 logger.warning("MSTeams activity handling failed: {}", e)
170
171 self.send_response(200)
172 self.send_header("Content-Type", "application/json")
173 self.end_headers()
174 self.wfile.write(b"{}")
175
176 def log_message(self, format: str, *args: Any) -> None:
177 return
178
179 self._server = ThreadingHTTPServer((self.config.host, self.config.port), Handler)
180 self._server_thread = threading.Thread(
181 target=self._server.serve_forever,
182 name="nanobot-msteams",
183 daemon=True,
184 )
185 self._server_thread.start()
186
187 logger.info(
188 "MSTeams webhook listening on http://{}:{}{}",
189 self.config.host,
190 self.config.port,
191 self.config.path,
192 )
193
194 while self._running:
195 await asyncio.sleep(1)
196
197 async def stop(self) -> None:
198 """Stop the channel."""
199 self._running = False
200 if self._server:
201 self._server.shutdown()
202 self._server.server_close()
203 self._server = None
204 if self._server_thread and self._server_thread.is_alive():
205 self._server_thread.join(timeout=2)
206 self._server_thread = None
207 if self._http:
208 await self._http.aclose()
209 self._http = None
210
211 async def send(self, msg: OutboundMessage) -> None:
212 """Send a plain text reply into an existing Teams conversation."""
213 if not self._http:
214 raise RuntimeError("MSTeams HTTP client not initialized")
215
216 ref = self._conversation_refs.get(str(msg.chat_id))
217 if not ref:
218 raise RuntimeError(f"MSTeams conversation ref not found for chat_id={msg.chat_id}")
219
220 token = await self._get_access_token()
221 base_url = f"{ref.service_url.rstrip('/')}/v3/conversations/{ref.conversation_id}/activities"
222 use_thread_reply = self.config.reply_in_thread and bool(ref.activity_id)
223 url = f"{base_url}/{ref.activity_id}" if use_thread_reply else base_url
224 headers = {
225 "Authorization": f"Bearer {token}",
226 "Content-Type": "application/json",
227 }
228 payload = {
229 "type": "message",
230 "text": msg.content or " ",
231 }
232 if use_thread_reply:
233 payload["replyToId"] = ref.activity_id
234
235 try:
236 resp = await self._http.post(url, headers=headers, json=payload)
237 resp.raise_for_status()
238 logger.info("MSTeams message sent to {}", ref.conversation_id)
239 except Exception as e:
240 logger.error("MSTeams send failed: {}", e)
241 raise
242
243 async def _handle_activity(self, activity: dict[str, Any]) -> None:
244 """Handle inbound Teams/Bot Framework activity."""
245 if activity.get("type") != "message":
246 return
247
248 conversation = activity.get("conversation") or {}
249 from_user = activity.get("from") or {}
250 recipient = activity.get("recipient") or {}
251 channel_data = activity.get("channelData") or {}
252
253 sender_id = str(from_user.get("aadObjectId") or from_user.get("id") or "").strip()
254 conversation_id = str(conversation.get("id") or "").strip()
255 service_url = str(activity.get("serviceUrl") or "").strip()
256 activity_id = str(activity.get("id") or "").strip()
257 conversation_type = str(conversation.get("conversationType") or "").strip()
258
259 if not sender_id or not conversation_id or not service_url:
260 return
261
262 if recipient.get("id") and from_user.get("id") == recipient.get("id"):
263 return
264
265 # DM-only MVP: ignore group/channel traffic for now
266 if conversation_type and conversation_type not in ("personal", ""):
267 logger.debug("MSTeams ignoring non-DM conversation {}", conversation_type)
268 return
269
270 text = self._sanitize_inbound_text(activity)
271 if not text:
272 text = self.config.mention_only_response.strip()
273 if not text:
274 logger.debug("MSTeams ignoring empty message after Teams text sanitization")
275 return
276
277 if not self.is_allowed(sender_id):
278 logger.warning(
279 "Access denied for sender {} on channel {}. "
280 "Add them to allowFrom list in config to grant access.",
281 sender_id, self.name,
282 )
283 return
284
285 self._conversation_refs[conversation_id] = ConversationRef(
286 service_url=service_url,
287 conversation_id=conversation_id,
288 bot_id=str(recipient.get("id") or "") or None,
289 activity_id=activity_id or None,
290 conversation_type=conversation_type or None,
291 tenant_id=str((channel_data.get("tenant") or {}).get("id") or "") or None,
292 )
293 self._save_refs()
294
295 await self._handle_message(
296 sender_id=sender_id,
297 chat_id=conversation_id,
298 content=text,
299 metadata={
300 "msteams": {
301 "activity_id": activity_id,
302 "conversation_id": conversation_id,
303 "conversation_type": conversation_type or "personal",
304 "from_name": from_user.get("name"),
305 }
306 },
307 )
308
309 def _sanitize_inbound_text(self, activity: dict[str, Any]) -> str:
310 """Extract the user-authored text from a Teams activity."""
311 text = str(activity.get("text") or "")
312 text = self._strip_possible_bot_mention(text)
313
314 channel_data = activity.get("channelData") or {}
315 reply_to_id = str(activity.get("replyToId") or "").strip()
316 normalized_preview = html.unescape(text).replace("&rsquo", "’").strip()
317 normalized_preview = normalized_preview.replace("\r\n", "\n").replace("\r", "\n")
318 preview_lines = [line.strip() for line in normalized_preview.split("\n")]
319 while preview_lines and not preview_lines[0]:
320 preview_lines.pop(0)
321 first_line = preview_lines[0] if preview_lines else ""
322 looks_like_quote_wrapper = first_line.lower().startswith("replying to ") or first_line.startswith("Reply wrapper")
323
324 if reply_to_id or channel_data.get("messageType") == "reply" or looks_like_quote_wrapper:
325 text = self._normalize_teams_reply_quote(text)
326
327 return text.strip()
328
329 def _strip_possible_bot_mention(self, text: str) -> str:
330 """Remove simple Teams mention markup from message text."""
331 cleaned = re.sub(r"<at\b[^>]*>.*?</at>", " ", text, flags=re.IGNORECASE | re.DOTALL)
332 cleaned = re.sub(r"[^\S\r\n]+", " ", cleaned)
333 cleaned = re.sub(r"(?:\r?\n){3,}", "\n\n", cleaned)
334 return cleaned.strip()
335
336 def _normalize_teams_reply_quote(self, text: str) -> str:
337 """Normalize Teams quoted replies into a compact structured form."""
338 cleaned = html.unescape(text).replace("&rsquo", "’").strip()
339 if not cleaned:
340 return ""
341
342 normalized_newlines = cleaned.replace("\r\n", "\n").replace("\r", "\n")
343 lines = [line.strip() for line in normalized_newlines.split("\n")]
344 while lines and not lines[0]:
345 lines.pop(0)
346
347 # Observed native Teams reply wrapper:
348 # Replying to Bob Smith
349 # actual reply text
350 if len(lines) >= 2 and lines[0].lower().startswith("replying to "):
351 quoted = lines[0][len("replying to ") :].strip(" :")
352 reply = "\n".join(lines[1:]).strip()
353 return self._format_reply_with_quote(quoted, reply)
354
355 # Observed reply wrapper where the quoted content is surfaced after a
356 # synthetic "Reply wrapper" header, sometimes with a blank line separating quote
357 # and reply, and sometimes as a compact line-based fallback shape.
358 if lines and lines[0].strip().startswith("Reply wrapper"):
359 body = normalized_newlines.split("\n", 1)[1] if "\n" in normalized_newlines else ""
360 body = body.lstrip()
361 parts = re.split(r"\n\s*\n", body, maxsplit=1)
362 if len(parts) == 2:
363 quoted = re.sub(r"\s+", " ", parts[0]).strip()
364 reply = re.sub(r"\s+", " ", parts[1]).strip()
365 if quoted or reply:
366 return self._format_reply_with_quote(quoted, reply)
367
368 body_lines = [line.strip() for line in body.split("\n") if line.strip()]
369 if body_lines:
370 quoted = " ".join(body_lines[:-1]).strip()
371 reply = body_lines[-1].strip()
372 if quoted and reply:
373 return self._format_reply_with_quote(quoted, reply)
374
375 # Observed compact fallback where the relay flattens quote and reply into
376 # a single line after the synthetic Reply wrapper prefix.
377 compact = re.sub(r"\s+", " ", normalized_newlines).strip()
378 if compact.startswith("Reply wrapper "):
379 compact = compact[len("Reply wrapper ") :].strip()
380 for boundary in (". ", "! ", "? ", "… "):
381 idx = compact.rfind(boundary)
382 if idx == -1:
383 continue
384 quoted = compact[: idx + 1].strip()
385 reply = compact[idx + len(boundary) :].strip()
386 if quoted and reply and len(reply) <= 160:
387 return self._format_reply_with_quote(quoted, reply)
388
389 return cleaned
390
391 def _format_reply_with_quote(self, quoted: str, reply: str) -> str:
392 """Format a reply-with-context message for the model without Teams wrapper noise."""
393 quoted = quoted.strip()
394 reply = reply.strip()
395 if quoted and reply:
396 return f"User is replying to: {quoted}\nUser reply: {reply}"
397 if reply:
398 return reply
399 return quoted
400
401 async def _validate_inbound_auth(self, auth_header: str, activity: dict[str, Any]) -> None:
402 """Validate inbound Bot Framework bearer token."""
403 if not MSTEAMS_AVAILABLE:
404 raise RuntimeError("PyJWT not installed. Run: pip install echo-director-agent[msteams]")
405
406 if not auth_header.lower().startswith("bearer "):
407 raise ValueError("missing bearer token")
408
409 token = auth_header.split(" ", 1)[1].strip()
410 if not token:
411 raise ValueError("empty bearer token")
412
413 header = jwt.get_unverified_header(token)
414 kid = str(header.get("kid") or "").strip()
415 if not kid:
416 raise ValueError("missing token kid")
417
418 jwks = await self._get_botframework_jwks()
419 keys = jwks.get("keys") or []
420 jwk = next((key for key in keys if key.get("kid") == kid), None)
421 if not jwk:
422 raise ValueError(f"signing key not found for kid={kid}")
423
424 public_key = jwt.algorithms.RSAAlgorithm.from_jwk(json.dumps(jwk))
425 claims = jwt.decode(
426 token,
427 key=public_key,
428 algorithms=["RS256"],
429 audience=self.config.app_id,
430 issuer="https://api.botframework.com",
431 options={
432 "require": ["exp", "nbf", "iss", "aud"],
433 },
434 )
435
436 claim_service_url = str(
437 claims.get("serviceurl") or claims.get("serviceUrl") or "",
438 ).strip()
439 activity_service_url = str(activity.get("serviceUrl") or "").strip()
440 if claim_service_url and activity_service_url and claim_service_url != activity_service_url:
441 raise ValueError("serviceUrl claim mismatch")
442
443 async def _get_botframework_openid_config(self) -> dict[str, Any]:
444 """Fetch and cache Bot Framework OpenID configuration."""
445
446 now = time.time()
447 if self._botframework_openid_config and now < self._botframework_openid_config_expires_at:
448 return self._botframework_openid_config
449
450 if not self._http:
451 raise RuntimeError("MSTeams HTTP client not initialized")
452
453 resp = await self._http.get(self._botframework_openid_config_url)
454 resp.raise_for_status()
455 self._botframework_openid_config = resp.json()
456 self._botframework_openid_config_expires_at = now + 3600
457 return self._botframework_openid_config
458
459 async def _get_botframework_jwks(self) -> dict[str, Any]:
460 """Fetch and cache Bot Framework JWKS."""
461
462 now = time.time()
463 if self._botframework_jwks and now < self._botframework_jwks_expires_at:
464 return self._botframework_jwks
465
466 if not self._http:
467 raise RuntimeError("MSTeams HTTP client not initialized")
468
469 openid_config = await self._get_botframework_openid_config()
470 jwks_uri = str(openid_config.get("jwks_uri") or "").strip()
471 if not jwks_uri:
472 raise RuntimeError("Bot Framework OpenID config missing jwks_uri")
473
474 resp = await self._http.get(jwks_uri)
475 resp.raise_for_status()
476 self._botframework_jwks = resp.json()
477 self._botframework_jwks_expires_at = now + 3600
478 return self._botframework_jwks
479
480 def _load_refs(self) -> dict[str, ConversationRef]:
481 """Load stored conversation references."""
482 if not self._refs_path.exists():
483 return {}
484 try:
485 data = json.loads(self._refs_path.read_text(encoding="utf-8"))
486 out: dict[str, ConversationRef] = {}
487 for key, value in data.items():
488 out[key] = ConversationRef(**value)
489 return out
490 except Exception as e:
491 logger.warning("Failed to load MSTeams conversation refs: {}", e)
492 return {}
493
494 def _save_refs(self) -> None:
495 """Persist conversation references."""
496 try:
497 data = {
498 key: {
499 "service_url": ref.service_url,
500 "conversation_id": ref.conversation_id,
501 "bot_id": ref.bot_id,
502 "activity_id": ref.activity_id,
503 "conversation_type": ref.conversation_type,
504 "tenant_id": ref.tenant_id,
505 }
506 for key, ref in self._conversation_refs.items()
507 }
508 self._refs_path.write_text(json.dumps(data, indent=2), encoding="utf-8")
509 except Exception as e:
510 logger.warning("Failed to save MSTeams conversation refs: {}", e)
511
512 async def _get_access_token(self) -> str:
513 """Fetch an access token for Bot Framework / Azure Bot auth."""
514
515 now = time.time()
516 if self._token and now < self._token_expires_at - 60:
517 return self._token
518
519 if not self._http:
520 raise RuntimeError("MSTeams HTTP client not initialized")
521
522 tenant = (self.config.tenant_id or "").strip() or "botframework.com"
523 token_url = f"https://login.microsoftonline.com/{tenant}/oauth2/v2.0/token"
524 data = {
525 "grant_type": "client_credentials",
526 "client_id": self.config.app_id,
527 "client_secret": self.config.app_password,
528 "scope": "https://api.botframework.com/.default",
529 }
530 resp = await self._http.post(token_url, data=data)
531 resp.raise_for_status()
532 payload = resp.json()
533 self._token = payload["access_token"]
534 self._token_expires_at = now + int(payload.get("expires_in", 3600))
535 return self._token
536
536 lines PYTHON