"""Microsoft Teams (send/read chats) in-process MCP for Jarvis. Gives Claude real Teams hands — message people, list recent chats, read a conversation — as Ahmed himself, via Microsoft Graph with DELEGATED permissions (acts on his behalf, in his name). No bot, no metered API (Teams Graph APIs stopped being metered in Aug 2025). Auth: MSAL device-code flow against an Azure Entra app registration Ahmed (as tenant admin) creates once. Public client, NO client secret. The first Teams action prints a short code + URL (microsoft.com/devicelogin); enter it once and the token caches to ``teams_token_cache.json`` and refreshes silently forever after. Runs IN-PROCESS via the Agent SDK MCP server (key ``"teams"``; Claude sees ``mcp__teams__``). ``msal`` is imported lazily so importing this module never hard-fails when it's missing. Azure setup (one-time, Ahmed as admin — see README): 1. Entra ID → App registrations → New. Note the Application (client) ID and Directory (tenant) ID. 2. Authentication → Allow public client flows = Yes (enables device code). 3. API permissions → Microsoft Graph → Delegated: Chat.ReadWrite, ChatMessage.Send, User.ReadBasic.All, User.Read → Grant admin consent. Config (``.env`` / env): TEAMS_CLIENT_ID the Application (client) ID (REQUIRED — dark without it) TEAMS_TENANT_ID Directory (tenant) ID, or "organizations" (default) Set ``MCP_TEAMS=0`` to disable. """ from __future__ import annotations import json import os import urllib.error import urllib.parse import urllib.request from pathlib import Path _ROOT = Path(__file__).resolve().parent.parent _GRAPH = "https://graph.microsoft.com/v1.0" _SCOPES = ["Chat.ReadWrite", "ChatMessage.Send", "User.ReadBasic.All", "User.Read"] _CACHE = _ROOT / "teams_token_cache.json" _APP = None # cached MSAL app _ME = None # cached {"id":..., "displayName":...} def enabled() -> bool: """Whether the Teams MCP should be exposed to Claude. Dark until the Azure app's client id is configured.""" if os.environ.get("MCP_TEAMS", "1") == "0": return False return bool(os.environ.get("TEAMS_CLIENT_ID", "").strip()) def _text(msg: str) -> dict: return {"content": [{"type": "text", "text": msg}]} def _strip_html(s: str) -> str: import re return re.sub(r"<[^>]+>", "", s or "").replace(" ", " ").strip() # --------------------------------------------------------------------------- # auth (blocking — called via asyncio.to_thread) # --------------------------------------------------------------------------- def _msal_app(): global _APP if _APP is not None: return _APP import msal cache = msal.SerializableTokenCache() if _CACHE.exists(): cache.deserialize(_CACHE.read_text()) tenant = os.environ.get("TEAMS_TENANT_ID", "organizations").strip() \ or "organizations" _APP = msal.PublicClientApplication( os.environ["TEAMS_CLIENT_ID"].strip(), authority=f"https://login.microsoftonline.com/{tenant}", token_cache=cache, ) _APP._cv_cache = cache # keep a handle for persisting after acquire return _APP def _persist(app) -> None: cache = getattr(app, "_cv_cache", None) if cache is not None and cache.has_state_changed: _CACHE.write_text(cache.serialize()) def _token() -> str: """Return a valid Graph access token, running the one-time device-code sign-in if there's no cached account. Blocking.""" app = _msal_app() result = None accounts = app.get_accounts() if accounts: result = app.acquire_token_silent(_SCOPES, account=accounts[0]) if not result: flow = app.initiate_device_flow(scopes=_SCOPES) if "user_code" not in flow: raise RuntimeError( f"device flow failed: {flow.get('error_description', flow)}") # This message is how Ahmed completes first-time sign-in. print("\n [teams] ONE-TIME SIGN-IN NEEDED:\n " + flow["message"] + "\n") result = app.acquire_token_by_device_flow(flow) # blocks until done _persist(app) if "access_token" not in result: raise RuntimeError( f"token error: {result.get('error_description', result)}") return result["access_token"] # --------------------------------------------------------------------------- # Graph REST (blocking) # --------------------------------------------------------------------------- def _graph(method: str, path: str, body: dict | None = None) -> dict: url = path if path.startswith("http") else _GRAPH + path data = json.dumps(body).encode() if body is not None else None req = urllib.request.Request(url, data=data, method=method) req.add_header("Authorization", f"Bearer {_token()}") req.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(req, timeout=30) as r: raw = r.read() return json.loads(raw) if raw else {} except urllib.error.HTTPError as e: detail = e.read().decode("utf-8", "replace")[:400] raise RuntimeError(f"Graph {method} {path} -> {e.code}: {detail}") def _me() -> dict: global _ME if _ME is None: m = _graph("GET", "/me") _ME = {"id": m.get("id"), "displayName": m.get("displayName", "me")} return _ME def _find_user(name: str) -> dict | None: """Resolve a person by display name or email to {id, displayName, mail}.""" name = name.strip() if "@" in name: # looks like an email try: u = _graph("GET", f"/users/{urllib.parse.quote(name)}") return {"id": u["id"], "displayName": u.get("displayName", name), "mail": u.get("mail") or name} except Exception: # noqa: BLE001 — fall through to name search pass q = urllib.parse.quote(f"startswith(displayName,'{name}')") res = _graph("GET", f"/users?$filter={q}&$top=5" "&$select=id,displayName,mail") users = res.get("value", []) if not users: return None u = users[0] return {"id": u["id"], "displayName": u.get("displayName", name), "mail": u.get("mail")} def _one_on_one_chat(user_id: str) -> str: """Get (or create — Graph dedups) the 1:1 chat id with a user.""" me = _me() body = { "chatType": "oneOnOne", "members": [ {"@odata.type": "#microsoft.graph.aadUserConversationMember", "roles": ["owner"], "user@odata.bind": f"https://graph.microsoft.com/v1.0/users('{me['id']}')"}, {"@odata.type": "#microsoft.graph.aadUserConversationMember", "roles": ["owner"], "user@odata.bind": f"https://graph.microsoft.com/v1.0/users('{user_id}')"}, ], } return _graph("POST", "/chats", body)["id"] def _send_chat(person: str, message: str) -> str: u = _find_user(person) if not u: return f"Couldn't find anyone named '{person}' in your organization." chat_id = _one_on_one_chat(u["id"]) _graph("POST", f"/chats/{chat_id}/messages", {"body": {"content": message}}) return f"Messaged {u['displayName']} on Teams." def _list_chats(limit: int) -> str: res = _graph("GET", f"/me/chats?$top={limit}&$expand=members" "&$orderby=lastMessagePreview/createdDateTime desc") chats = res.get("value", []) if not chats: return "No recent Teams chats." my_id = _me()["id"] lines = [] for c in chats: if c.get("chatType") == "oneOnOne": others = [m.get("displayName", "?") for m in c.get("members", []) if m.get("userId") != my_id] name = ", ".join(others) or "(unknown)" else: name = c.get("topic") or "(group chat)" prev = _strip_html( (c.get("lastMessagePreview") or {}).get("body", {}).get( "content", ""))[:80] lines.append(f"[{c['id']}] {name}" + (f" — {prev}" if prev else "")) return "\n".join(lines) def _read_chat(person: str, count: int) -> str: """Read the last `count` messages of the 1:1 chat with a person.""" u = _find_user(person) if not u: return f"Couldn't find anyone named '{person}'." # find an existing chat with them (don't create just to read) res = _graph("GET", "/me/chats?$top=50&$expand=members") my_id = _me()["id"] chat_id = None for c in res.get("value", []): if c.get("chatType") != "oneOnOne": continue if any(m.get("userId") == u["id"] for m in c.get("members", [])): chat_id = c["id"] break if not chat_id: return f"No existing Teams chat with {u['displayName']}." msgs = _graph("GET", f"/chats/{chat_id}/messages?$top={count}" ).get("value", []) if not msgs: return f"No messages in the chat with {u['displayName']}." lines = [] for m in reversed(msgs): # oldest first who = (m.get("from") or {}).get("user", {}).get( "displayName") or "system" text = _strip_html(m.get("body", {}).get("content", "")) if text: lines.append(f"{who}: {text}") return f"Chat with {u['displayName']}:\n" + "\n".join(lines) def build_server(): """Build and return the in-process ``teams`` MCP server.""" import asyncio from claude_agent_sdk import tool, create_sdk_mcp_server @tool("teams_send", "Send a Microsoft Teams message to a person by name or email. Finds " "them in Ahmed's organization and messages their 1:1 chat, as Ahmed.", {"person": str, "message": str}) async def teams_send(args: dict) -> dict: person = str(args.get("person", "")).strip() message = str(args.get("message", "")).strip() if not person or not message: return _text("teams_send failed: need both person and message.") try: return _text(await asyncio.to_thread(_send_chat, person, message)) except Exception as e: # noqa: BLE001 return _text(f"teams_send failed: {e}") @tool("teams_list_chats", "List Ahmed's most recent Teams chats (people/groups) with a chat id " "and last-message preview for each.", {"limit": int}) async def teams_list_chats(args: dict) -> dict: try: limit = max(1, min(int(args.get("limit") or 15), 30)) return _text(await asyncio.to_thread(_list_chats, limit)) except Exception as e: # noqa: BLE001 return _text(f"teams_list_chats failed: {e}") @tool("teams_read_chat", "Read the recent messages of Ahmed's 1:1 Teams chat with a person " "(by name or email).", {"person": str, "count": int}) async def teams_read_chat(args: dict) -> dict: person = str(args.get("person", "")).strip() if not person: return _text("teams_read_chat failed: no person.") try: count = max(1, min(int(args.get("count") or 10), 30)) return _text(await asyncio.to_thread(_read_chat, person, count)) except Exception as e: # noqa: BLE001 return _text(f"teams_read_chat failed: {e}") return create_sdk_mcp_server( name="teams", version="1.0.0", tools=[teams_send, teams_list_chats, teams_read_chat], )