"""Teams operations for Microsoft Graph API."""
import base64
import os
import logging
from typing import Dict, List, Any
from .base import BaseOperation, OperationError
from ..graph_client import MicrosoftGraphClient
logger = logging.getLogger(__name__)
class TeamsOperations(BaseOperation):
"""Operations for Microsoft Teams management."""
def get_supported_actions(self) -> List[str]:
"""Return supported Teams actions."""
return [
"list_chats", "get_chat", "create_chat", "send_message",
"list_teams", "list_channels", "send_channel_message"
]
def _validate_action_params(self, action: str, params: Dict) -> None:
"""Validate parameters for Teams actions."""
# access_token is required for all actions (from TrustyVault)
self._require_param(params, "access_token", str)
self._require_param(params, "microsoft_user", str)
if action == "get_chat":
self._require_param(params, "chat_id", str)
elif action == "send_message":
self._require_param(params, "message", str)
# Either chat_id OR recipient_email required (not both necessary)
chat_id = params.get("chat_id")
recipient_email = params.get("recipient_email")
if not chat_id and not recipient_email:
raise OperationError(
code="INVALID_PARAM",
message="Either 'chat_id' or 'recipient_email' is required"
)
elif action == "create_chat":
members = self._require_param(params, "members", list)
if len(members) == 0:
raise OperationError(
code="INVALID_PARAM",
message="At least one member required"
)
for email in members:
self._validate_email(email)
elif action == "list_channels":
self._require_param(params, "team_id", str)
elif action == "send_channel_message":
self._require_param(params, "team_id", str)
self._require_param(params, "channel_id", str)
self._require_param(params, "message", str)
def _execute_action(self, action: str, params: Dict) -> Any:
"""Execute Teams action."""
if action == "list_chats":
return self._action_list_chats(params)
elif action == "get_chat":
return self._action_get_chat(params)
elif action == "create_chat":
return self._action_create_chat(params)
elif action == "send_message":
return self._action_send_message(params)
elif action == "list_teams":
return self._action_list_teams(params)
elif action == "list_channels":
return self._action_list_channels(params)
elif action == "send_channel_message":
return self._action_send_channel_message(params)
def _action_list_chats(self, params: Dict) -> Dict:
"""List user's chats with members information."""
access_token = params["access_token"]
microsoft_user = params["microsoft_user"]
try:
# Get basic chat list with $expand to include members
# Create GraphClient with access_token from TrustyVault
client = MicrosoftGraphClient(access_token=access_token)
response = client.get(
f"/users/{microsoft_user}/chats?$expand=members"
)
chats = response.get("value", [])
# Log first chat to debug member structure
if chats:
logger.info(f"First chat sample (with members): {chats[0]}")
return {"chats": chats, "count": len(chats)}
except Exception as e:
raise OperationError(
code="LIST_CHATS_FAILED",
message=f"Failed to list chats: {str(e)}"
)
def _action_get_chat(self, params: Dict) -> Dict:
"""Get chat details."""
access_token = params["access_token"]
microsoft_user = params["microsoft_user"]
chat_id = params["chat_id"]
try:
# Create GraphClient with access_token from TrustyVault
client = MicrosoftGraphClient(access_token=access_token)
chat = client.get(f"/chats/{chat_id}")
return {"chat": chat}
except Exception as e:
raise OperationError(
code="GET_CHAT_FAILED",
message=f"Failed to get chat: {str(e)}",
details={"chat_id": chat_id}
)
def _resolve_user_upn(self, email_or_name: str, access_token: str) -> str:
"""
Resolve user email/name to UPN (User Principal Name).
Microsoft Graph API accepts both UPN and email in /users/{id} endpoint,
but for Teams chat members we need the actual UPN for the user@odata.bind format.
Args:
email_or_name: Email address, UPN, or display name to resolve
access_token: Microsoft Graph API access token
Returns:
UPN (User Principal Name) or original input if resolution fails
"""
# Create GraphClient with access_token
client = MicrosoftGraphClient(access_token=access_token)
# First, try direct lookup (works for both UPN and email if they match)
try:
user_response = client.get(f"/users/{email_or_name}")
upn = user_response.get("userPrincipalName")
if upn:
logger.info(f"✅ Resolved {email_or_name} → {upn} (direct lookup)")
return upn
except Exception as e:
logger.debug(f"Direct lookup failed for {email_or_name}: {e}")
# Try searching by email or UPN
try:
search_response = client.get(
f"/users?$filter=mail eq '{email_or_name}' or userPrincipalName eq '{email_or_name}'"
)
users = search_response.get("value", [])
if users:
upn = users[0].get("userPrincipalName", email_or_name)
logger.info(f"✅ Resolved {email_or_name} → {upn} (filter search)")
return upn
# If still not found, try display name search
# Extract name from email if it looks like firstname.lastname@domain
if "@" in email_or_name:
name_part = email_or_name.split("@")[0].replace(".", " ")
search_response = client.get(
f"/users?$search=\"displayName:{name_part}\"",
headers={"ConsistencyLevel": "eventual"}
)
users = search_response.get("value", [])
if users:
upn = users[0].get("userPrincipalName", email_or_name)
logger.info(f"✅ Resolved {email_or_name} → {upn} (displayName search)")
return upn
except Exception as e:
logger.warning(f"Search failed for {email_or_name}: {e}")
# If all fails, return original (might still work if it's already a valid UPN)
logger.warning(f"⚠️ Could not resolve {email_or_name}, using as-is")
return email_or_name
def _action_create_chat(self, params: Dict) -> Dict:
"""Create new chat.
Accepts email addresses in 'members' parameter and resolves them to UPNs.
Automatically includes the authenticated user (microsoft_user) as owner.
"""
access_token = params["access_token"]
microsoft_user = params["microsoft_user"]
members = params["members"]
topic = params.get("topic")
# Resolve all member emails to UPNs
resolved_members = []
for member in members:
resolved_upn = self._resolve_user_upn(member, access_token)
resolved_members.append(resolved_upn)
# Always include the caller (microsoft_user) in the chat members if not already present
all_members = list(resolved_members)
if microsoft_user not in all_members:
all_members.insert(0, microsoft_user) # Add caller as first member
# Build chat request
chat_request = {
"chatType": "group" if len(all_members) > 2 else "oneOnOne",
"members": [
{
"@odata.type": "#microsoft.graph.aadUserConversationMember",
"roles": ["owner"],
"user@odata.bind": f"https://graph.microsoft.com/v1.0/users('{email}')"
}
for email in all_members
]
}
if topic and len(members) > 1:
chat_request["topic"] = topic
try:
# Create GraphClient with access_token from TrustyVault
client = MicrosoftGraphClient(access_token=access_token)
chat = client.post("/chats", json=chat_request)
return {"chat": chat, "chat_id": chat.get("id"), "created": True}
except Exception as e:
error_msg = str(e).lower()
# If chat already exists (1-on-1 chat duplicate), try to find it
if "already exists" in error_msg or "duplicate" in error_msg or len(all_members) == 2:
try:
# Get all chats and find the one with these members
chats = client.get("/chats")
for chat in chats.get("value", []):
if chat.get("chatType") != "oneOnOne":
continue
# Get members of this chat
chat_id = chat["id"]
members_response = client.get(
f"/chats/{chat_id}/members"
)
chat_member_emails = {
m.get("email", "").lower()
for m in members_response.get("value", [])
if m.get("email")
}
target_emails = {email.lower() for email in all_members}
# Check if this chat has exactly our members
if chat_member_emails == target_emails:
return {"chat": chat, "chat_id": chat_id, "created": False, "found_existing": True}
# If we get here, couldn't find existing chat - raise original error
raise OperationError(
code="CREATE_CHAT_FAILED",
message=f"Failed to create chat and couldn't find existing: {str(e)}",
details={"members": members}
)
except Exception as search_error:
# If search fails, raise original create error
raise OperationError(
code="CREATE_CHAT_FAILED",
message=f"Failed to create chat: {str(e)}. Search for existing also failed: {str(search_error)}",
details={"members": members}
)
else:
# Not a duplicate error, raise original
raise OperationError(
code="CREATE_CHAT_FAILED",
message=f"Failed to create chat: {str(e)}",
details={"members": members}
)
def _get_or_create_chat(self, access_token: str, microsoft_user: str, recipient_email: str) -> str:
"""Find existing 1-on-1 chat or create new one.
Args:
access_token: Microsoft Graph API token
microsoft_user: Authenticated user UPN
recipient_email: Email (display or UPN) of the other person
Returns:
chat_id: ID of existing or newly created chat
"""
client = MicrosoftGraphClient(access_token=access_token)
# Resolve recipient email to UPN if needed
recipient_upn = self._resolve_user_upn(recipient_email, access_token)
# Check if trying to message yourself - Microsoft Teams doesn't allow 1-on-1 chat with yourself
if recipient_upn.lower() == microsoft_user.lower():
raise OperationError(
code="SELF_MESSAGE_NOT_ALLOWED",
message=f"Cannot send Teams message to yourself (Microsoft Teams limitation). Use email_send_message tool instead.",
details={"microsoft_user": microsoft_user, "recipient": recipient_email}
)
# Step 1: Try to find existing 1-on-1 chat
try:
chats_response = client.get(f"/users/{microsoft_user}/chats?$expand=members")
chats = chats_response.get("value", [])
for chat in chats:
if chat.get("chatType") != "oneOnOne":
continue
# Check if this chat has exactly 2 members: caller + recipient
members = chat.get("members", [])
if len(members) != 2:
continue
member_upns = {
m.get("email", "").lower()
for m in members
if m.get("email")
}
target_upns = {microsoft_user.lower(), recipient_upn.lower()}
if member_upns == target_upns:
# Found existing chat!
return chat["id"]
except Exception as e:
logger.warning(f"Failed to search existing chats: {e}. Will create new chat.")
# Step 2: No existing chat found - create new one
try:
create_result = self._action_create_chat({
"access_token": access_token,
"microsoft_user": microsoft_user,
"members": [recipient_email] # Pass original email (will be resolved by create_chat)
})
return create_result["chat_id"]
except Exception as e:
raise OperationError(
code="CHAT_DISCOVERY_FAILED",
message=f"Failed to find or create chat with {recipient_email}: {str(e)}",
details={"recipient": recipient_email}
)
def _action_send_message(self, params: Dict) -> Dict:
"""Send message to chat with optional file attachments.
Supports two modes:
1. Direct chat_id: Send to existing chat immediately
2. recipient_email: Auto-discover or create 1-on-1 chat, then send
"""
access_token = params["access_token"]
microsoft_user = params["microsoft_user"]
message = params["message"]
content_type = params.get("content_type", "text")
# Mode 1: chat_id provided - use directly
chat_id = params.get("chat_id")
# Mode 2: recipient_email provided - find or create chat
recipient_email = params.get("recipient_email")
if not chat_id and recipient_email:
# Auto-discover or create 1-on-1 chat
chat_id = self._get_or_create_chat(access_token, microsoft_user, recipient_email)
# At this point, chat_id is guaranteed to exist
# Create GraphClient
client = MicrosoftGraphClient(access_token=access_token)
message_data = {
"body": {
"contentType": content_type,
"content": message
}
}
# Handle attachments if provided
attachments = params.get("attachments", [])
if attachments:
message_data["attachments"] = []
attachment_markers = []
for attachment in attachments:
if "file_path" in attachment:
file_path = attachment["file_path"]
if not os.path.exists(file_path):
raise OperationError(
code="FILE_NOT_FOUND",
message=f"Attachment file not found: {file_path}"
)
file_name = attachment.get("name") or os.path.basename(file_path)
# Upload file to OneDrive and get sharing link
uploaded_file_url = self._upload_file_to_onedrive(
file_path, file_name, access_token, microsoft_user
)
# Generate unique ID for attachment
import uuid
attachment_id = str(uuid.uuid4())
# Add attachment as reference (required for Teams)
message_data["attachments"].append({
"id": attachment_id,
"contentType": "reference",
"contentUrl": uploaded_file_url,
"name": file_name
})
# Track marker to add to message body
attachment_markers.append(f'<attachment id="{attachment_id}"></attachment>')
# Add attachment markers to message body
if attachment_markers:
marker_html = "\n".join(attachment_markers)
if content_type == "html":
message_data["body"]["content"] += f"\n{marker_html}"
else:
# If text, convert to HTML to support attachment markers
message_data["body"]["contentType"] = "html"
message_data["body"]["content"] = f"<p>{message}</p>\n{marker_html}"
try:
sent_message = client.post(
f"/chats/{chat_id}/messages",
json=message_data
)
return {
"sent": True,
"message_id": sent_message.get("id"),
"chat_id": chat_id,
"attachments_count": len(attachments)
}
except Exception as e:
raise OperationError(
code="SEND_MESSAGE_FAILED",
message=f"Failed to send message: {str(e)}",
details={"chat_id": chat_id}
)
def _action_list_teams(self, params: Dict) -> Dict:
"""List user's teams."""
access_token = params["access_token"]
microsoft_user = params["microsoft_user"]
try:
# Create GraphClient with access_token from TrustyVault
client = MicrosoftGraphClient(access_token=access_token)
response = client.get(f"/users/{microsoft_user}/joinedTeams")
teams = response.get("value", [])
return {"teams": teams, "count": len(teams)}
except Exception as e:
raise OperationError(
code="LIST_TEAMS_FAILED",
message=f"Failed to list teams: {str(e)}"
)
def _action_list_channels(self, params: Dict) -> Dict:
"""List channels in team."""
access_token = params["access_token"]
microsoft_user = params["microsoft_user"]
team_id = params["team_id"]
try:
# Create GraphClient with access_token from TrustyVault
client = MicrosoftGraphClient(access_token=access_token)
response = client.get(f"/teams/{team_id}/channels")
channels = response.get("value", [])
return {"channels": channels, "count": len(channels), "team_id": team_id}
except Exception as e:
raise OperationError(
code="LIST_CHANNELS_FAILED",
message=f"Failed to list channels: {str(e)}",
details={"team_id": team_id}
)
def _action_send_channel_message(self, params: Dict) -> Dict:
"""Send message to channel."""
access_token = params["access_token"]
microsoft_user = params["microsoft_user"]
team_id = params["team_id"]
channel_id = params["channel_id"]
message = params["message"]
content_type = params.get("content_type", "html")
# Create GraphClient
client = MicrosoftGraphClient(access_token=access_token)
message_data = {
"body": {
"contentType": content_type,
"content": message
}
}
try:
sent_message = client.post(
f"/teams/{team_id}/channels/{channel_id}/messages",
json=message_data
)
return {
"sent": True,
"message_id": sent_message.get("id"),
"team_id": team_id,
"channel_id": channel_id
}
except Exception as e:
raise OperationError(
code="SEND_CHANNEL_MESSAGE_FAILED",
message=f"Failed to send channel message: {str(e)}",
details={"team_id": team_id, "channel_id": channel_id}
)
def _guess_content_type(self, filename: str) -> str:
"""Guess content type from filename extension."""
ext = filename.lower().split('.')[-1]
content_types = {
'pdf': 'application/pdf',
'doc': 'application/msword',
'docx': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document',
'xls': 'application/vnd.ms-excel',
'xlsx': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
'ppt': 'application/vnd.ms-powerpoint',
'pptx': 'application/vnd.openxmlformats-officedocument.presentationml.presentation',
'txt': 'text/plain',
'html': 'text/html',
'jpg': 'image/jpeg',
'jpeg': 'image/jpeg',
'png': 'image/png',
'gif': 'image/gif',
'zip': 'application/zip'
}
return content_types.get(ext, 'application/octet-stream')
def _upload_file_to_onedrive(
self,
file_path: str,
file_name: str,
access_token: str,
microsoft_user: str
) -> str:
"""Upload file to OneDrive and return sharing URL.
Args:
file_path: Path to the file to upload
file_name: Name of the file
access_token: Microsoft Graph API access token
microsoft_user: Microsoft user UPN (owner of the OneDrive)
Returns:
str: SharePoint/OneDrive URL that can be used as contentUrl for Teams attachment
"""
# Create GraphClient
client = MicrosoftGraphClient(access_token=access_token)
try:
# Read file content
with open(file_path, "rb") as f:
file_content = f.read()
# Upload to OneDrive - use unique filename to avoid conflicts
from datetime import datetime
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
unique_filename = f"{timestamp}_{file_name}"
# Use simple upload for files < 4MB
upload_response = client.put(
f"/users/{microsoft_user}/drive/root:/Attachments/{unique_filename}:/content",
data=file_content,
headers={"Content-Type": "application/octet-stream"}
)
# Get the item ID to create sharing link
item_id = upload_response.get("id")
if not item_id:
raise OperationError(
code="UPLOAD_FAILED",
message="Failed to get item ID from OneDrive upload response"
)
# Create a sharing link accessible to the organization
# This allows anyone in the organization to view the file
try:
sharing_response = client.post(
f"/users/{microsoft_user}/drive/items/{item_id}/createLink",
json={
"type": "view", # view permission (read-only)
"scope": "organization" # accessible to anyone in the organization
}
)
# Get the sharing URL from the response
sharing_link = sharing_response.get("link", {}).get("webUrl")
if sharing_link:
return sharing_link
else:
# Fallback to direct webUrl if sharing link creation fails
web_url = upload_response.get("webUrl")
if web_url:
return web_url
raise OperationError(
code="UPLOAD_FAILED",
message="Failed to get sharing link or webUrl from OneDrive"
)
except Exception as sharing_error:
# If sharing link creation fails, try using webUrl as fallback
web_url = upload_response.get("webUrl")
if web_url:
return web_url
raise OperationError(
code="SHARING_FAILED",
message=f"Failed to create sharing link: {str(sharing_error)}"
)
except Exception as e:
raise OperationError(
code="ONEDRIVE_UPLOAD_FAILED",
message=f"Failed to upload file to OneDrive: {str(e)}",
details={"file_name": file_name}
)
