"""
Google Chat MCP Tools
This module provides MCP tools for interacting with Google Chat API.
"""
import base64
import logging
import asyncio
from typing import Dict, List, Optional
import httpx
from googleapiclient.errors import HttpError
# Auth & server utilities
from auth.service_decorator import require_google_service, require_multiple_services
from core.server import server
from core.utils import handle_http_errors
logger = logging.getLogger(__name__)
# In-memory cache for user ID → display name (bounded to avoid unbounded growth)
_SENDER_CACHE_MAX_SIZE = 256
_sender_name_cache: Dict[str, str] = {}
def _cache_sender(user_id: str, name: str) -> None:
"""Store a resolved sender name, evicting oldest entries if cache is full."""
if len(_sender_name_cache) >= _SENDER_CACHE_MAX_SIZE:
to_remove = list(_sender_name_cache.keys())[: _SENDER_CACHE_MAX_SIZE // 2]
for k in to_remove:
del _sender_name_cache[k]
_sender_name_cache[user_id] = name
async def _resolve_sender(people_service, sender_obj: dict) -> str:
"""Resolve a Chat message sender to a display name.
Fast path: use displayName if the API already provided it.
Slow path: look up the user via the People API directory and cache the result.
"""
# Fast path — Chat API sometimes provides displayName directly
display_name = sender_obj.get("displayName")
if display_name:
return display_name
user_id = sender_obj.get("name", "") # e.g. "users/123456789"
if not user_id:
return "Unknown Sender"
# Check cache
if user_id in _sender_name_cache:
return _sender_name_cache[user_id]
# Try People API directory lookup
# Chat API uses "users/ID" but People API expects "people/ID"
people_resource = user_id.replace("users/", "people/", 1)
if people_service:
try:
person = await asyncio.to_thread(
people_service.people()
.get(resourceName=people_resource, personFields="names,emailAddresses")
.execute
)
names = person.get("names", [])
if names:
resolved = names[0].get("displayName", user_id)
_cache_sender(user_id, resolved)
return resolved
# Fall back to email if no name
emails = person.get("emailAddresses", [])
if emails:
resolved = emails[0].get("value", user_id)
_cache_sender(user_id, resolved)
return resolved
except HttpError as e:
logger.debug(f"People API lookup failed for {user_id}: {e}")
except Exception as e:
logger.debug(f"Unexpected error resolving {user_id}: {e}")
# Final fallback
_cache_sender(user_id, user_id)
return user_id
def _extract_rich_links(msg: dict) -> List[str]:
"""Extract URLs from RICH_LINK annotations (smart chips).
When a user pastes a Google Workspace URL in Chat and it renders as a
smart chip, the URL is NOT in the text field — it's only available in
the annotations array as a RICH_LINK with richLinkMetadata.uri.
"""
text = msg.get("text", "")
urls = []
for ann in msg.get("annotations", []):
if ann.get("type") == "RICH_LINK":
uri = ann.get("richLinkMetadata", {}).get("uri", "")
if uri and uri not in text:
urls.append(uri)
return urls
@server.tool()
@require_google_service("chat", "chat_spaces_readonly")
@handle_http_errors("list_spaces", service_type="chat")
async def list_spaces(
service,
user_google_email: str,
page_size: int = 100,
space_type: str = "all", # "all", "room", "dm"
) -> str:
"""
Lists Google Chat spaces (rooms and direct messages) accessible to the user.
Returns:
str: A formatted list of Google Chat spaces accessible to the user.
"""
logger.info(f"[list_spaces] Email={user_google_email}, Type={space_type}")
# Build filter based on space_type
filter_param = None
if space_type == "room":
filter_param = "spaceType = SPACE"
elif space_type == "dm":
filter_param = "spaceType = DIRECT_MESSAGE"
request_params = {"pageSize": page_size}
if filter_param:
request_params["filter"] = filter_param
response = await asyncio.to_thread(service.spaces().list(**request_params).execute)
spaces = response.get("spaces", [])
if not spaces:
return f"No Chat spaces found for type '{space_type}'."
output = [f"Found {len(spaces)} Chat spaces (type: {space_type}):"]
for space in spaces:
space_name = space.get("displayName", "Unnamed Space")
space_id = space.get("name", "")
space_type_actual = space.get("spaceType", "UNKNOWN")
output.append(f"- {space_name} (ID: {space_id}, Type: {space_type_actual})")
return "\n".join(output)
@server.tool()
@require_multiple_services(
[
{"service_type": "chat", "scopes": "chat_read", "param_name": "chat_service"},
{
"service_type": "people",
"scopes": "contacts_read",
"param_name": "people_service",
},
]
)
@handle_http_errors("get_messages", service_type="chat")
async def get_messages(
chat_service,
people_service,
user_google_email: str,
space_id: str,
page_size: int = 50,
order_by: str = "createTime desc",
) -> str:
"""
Retrieves messages from a Google Chat space.
Returns:
str: Formatted messages from the specified space.
"""
logger.info(f"[get_messages] Space ID: '{space_id}' for user '{user_google_email}'")
# Get space info first
space_info = await asyncio.to_thread(
chat_service.spaces().get(name=space_id).execute
)
space_name = space_info.get("displayName", "Unknown Space")
# Get messages
response = await asyncio.to_thread(
chat_service.spaces()
.messages()
.list(parent=space_id, pageSize=page_size, orderBy=order_by)
.execute
)
messages = response.get("messages", [])
if not messages:
return f"No messages found in space '{space_name}' (ID: {space_id})."
# Pre-resolve unique senders in parallel
sender_lookup = {}
for msg in messages:
s = msg.get("sender", {})
key = s.get("name", "")
if key and key not in sender_lookup:
sender_lookup[key] = s
resolved_names = await asyncio.gather(
*[_resolve_sender(people_service, s) for s in sender_lookup.values()]
)
sender_map = dict(zip(sender_lookup.keys(), resolved_names))
output = [f"Messages from '{space_name}' (ID: {space_id}):\n"]
for msg in messages:
sender_obj = msg.get("sender", {})
sender_key = sender_obj.get("name", "")
sender = sender_map.get(sender_key) or await _resolve_sender(
people_service, sender_obj
)
create_time = msg.get("createTime", "Unknown Time")
text_content = msg.get("text", "No text content")
msg_name = msg.get("name", "")
output.append(f"[{create_time}] {sender}:")
output.append(f" {text_content}")
rich_links = _extract_rich_links(msg)
for url in rich_links:
output.append(f" [linked: {url}]")
# Show attachments
attachments = msg.get("attachment", [])
for idx, att in enumerate(attachments):
att_name = att.get("contentName", "unnamed")
att_type = att.get("contentType", "unknown type")
att_resource = att.get("name", "")
output.append(f" [attachment {idx}: {att_name} ({att_type})]")
if att_resource:
output.append(
f" Use download_chat_attachment(message_id='{msg_name}', attachment_index={idx}) to download"
)
# Show thread info if this is a threaded reply
thread = msg.get("thread", {})
if msg.get("threadReply") and thread.get("name"):
output.append(f" [thread: {thread['name']}]")
# Show emoji reactions
reactions = msg.get("emojiReactionSummaries", [])
if reactions:
parts = []
for r in reactions:
emoji = r.get("emoji", {})
symbol = emoji.get("unicode", "")
if not symbol:
ce = emoji.get("customEmoji", {})
symbol = f":{ce.get('uid', '?')}:"
count = r.get("reactionCount", 0)
parts.append(f"{symbol}x{count}")
output.append(f" [reactions: {', '.join(parts)}]")
output.append(f" (Message ID: {msg_name})\n")
return "\n".join(output)
@server.tool()
@require_google_service("chat", "chat_write")
@handle_http_errors("send_message", service_type="chat")
async def send_message(
service,
user_google_email: str,
space_id: str,
message_text: str,
thread_key: Optional[str] = None,
thread_name: Optional[str] = None,
) -> str:
"""
Sends a message to a Google Chat space.
Args:
thread_name: Reply in an existing thread by its resource name (e.g. spaces/X/threads/Y).
thread_key: Reply in a thread by app-defined key (creates thread if not found).
Returns:
str: Confirmation message with sent message details.
"""
logger.info(f"[send_message] Email: '{user_google_email}', Space: '{space_id}'")
message_body = {"text": message_text}
request_params = {"parent": space_id, "body": message_body}
# Thread reply support
if thread_name:
message_body["thread"] = {"name": thread_name}
request_params["messageReplyOption"] = "REPLY_MESSAGE_FALLBACK_TO_NEW_THREAD"
elif thread_key:
message_body["thread"] = {"threadKey": thread_key}
request_params["messageReplyOption"] = "REPLY_MESSAGE_FALLBACK_TO_NEW_THREAD"
message = await asyncio.to_thread(
service.spaces().messages().create(**request_params).execute
)
message_name = message.get("name", "")
create_time = message.get("createTime", "")
msg = f"Message sent to space '{space_id}' by {user_google_email}. Message ID: {message_name}, Time: {create_time}"
logger.info(
f"Successfully sent message to space '{space_id}' by {user_google_email}"
)
return msg
@server.tool()
@require_multiple_services(
[
{"service_type": "chat", "scopes": "chat_read", "param_name": "chat_service"},
{
"service_type": "people",
"scopes": "contacts_read",
"param_name": "people_service",
},
]
)
@handle_http_errors("search_messages", service_type="chat")
async def search_messages(
chat_service,
people_service,
user_google_email: str,
query: str,
space_id: Optional[str] = None,
page_size: int = 25,
) -> str:
"""
Searches for messages in Google Chat spaces by text content.
Returns:
str: A formatted list of messages matching the search query.
"""
logger.info(f"[search_messages] Email={user_google_email}, Query='{query}'")
# If specific space provided, search within that space
if space_id:
response = await asyncio.to_thread(
chat_service.spaces()
.messages()
.list(parent=space_id, pageSize=page_size, filter=f'text:"{query}"')
.execute
)
messages = response.get("messages", [])
context = f"space '{space_id}'"
else:
# Search across all accessible spaces (this may require iterating through spaces)
# For simplicity, we'll search the user's spaces first
spaces_response = await asyncio.to_thread(
chat_service.spaces().list(pageSize=100).execute
)
spaces = spaces_response.get("spaces", [])
messages = []
for space in spaces[:10]: # Limit to first 10 spaces to avoid timeout
try:
space_messages = await asyncio.to_thread(
chat_service.spaces()
.messages()
.list(
parent=space.get("name"), pageSize=5, filter=f'text:"{query}"'
)
.execute
)
space_msgs = space_messages.get("messages", [])
for msg in space_msgs:
msg["_space_name"] = space.get("displayName", "Unknown")
messages.extend(space_msgs)
except HttpError as e:
logger.debug(
"Skipping space %s during search: %s", space.get("name"), e
)
continue
context = "all accessible spaces"
if not messages:
return f"No messages found matching '{query}' in {context}."
# Pre-resolve unique senders in parallel
sender_lookup = {}
for msg in messages:
s = msg.get("sender", {})
key = s.get("name", "")
if key and key not in sender_lookup:
sender_lookup[key] = s
resolved_names = await asyncio.gather(
*[_resolve_sender(people_service, s) for s in sender_lookup.values()]
)
sender_map = dict(zip(sender_lookup.keys(), resolved_names))
output = [f"Found {len(messages)} messages matching '{query}' in {context}:"]
for msg in messages:
sender_obj = msg.get("sender", {})
sender_key = sender_obj.get("name", "")
sender = sender_map.get(sender_key) or await _resolve_sender(
people_service, sender_obj
)
create_time = msg.get("createTime", "Unknown Time")
text_content = msg.get("text", "No text content")
space_name = msg.get("_space_name", "Unknown Space")
# Truncate long messages
if len(text_content) > 100:
text_content = text_content[:100] + "..."
rich_links = _extract_rich_links(msg)
links_suffix = "".join(f" [linked: {url}]" for url in rich_links)
attachments = msg.get("attachment", [])
att_suffix = "".join(
f" [attachment: {a.get('contentName', 'unnamed')} ({a.get('contentType', 'unknown type')})]"
for a in attachments
)
output.append(
f"- [{create_time}] {sender} in '{space_name}': {text_content}{links_suffix}{att_suffix}"
)
return "\n".join(output)
@server.tool()
@require_google_service("chat", "chat_write")
@handle_http_errors("create_reaction", service_type="chat")
async def create_reaction(
service,
user_google_email: str,
message_id: str,
emoji_unicode: str,
) -> str:
"""
Adds an emoji reaction to a Google Chat message.
Args:
message_id: The message resource name (e.g. spaces/X/messages/Y).
emoji_unicode: The emoji character to react with (e.g. 👍).
Returns:
str: Confirmation message.
"""
logger.info(f"[create_reaction] Message: '{message_id}', Emoji: '{emoji_unicode}'")
reaction = await asyncio.to_thread(
service.spaces()
.messages()
.reactions()
.create(
parent=message_id,
body={"emoji": {"unicode": emoji_unicode}},
)
.execute
)
reaction_name = reaction.get("name", "")
return f"Reacted with {emoji_unicode} on message {message_id}. Reaction ID: {reaction_name}"
@server.tool()
@handle_http_errors("download_chat_attachment", is_read_only=True, service_type="chat")
@require_google_service("chat", "chat_read")
async def download_chat_attachment(
service,
user_google_email: str,
message_id: str,
attachment_index: int = 0,
) -> str:
"""
Downloads an attachment from a Google Chat message and saves it to local disk.
In stdio mode, returns the local file path for direct access.
In HTTP mode, returns a temporary download URL (valid for 1 hour).
Args:
message_id: The message resource name (e.g. spaces/X/messages/Y).
attachment_index: Zero-based index of the attachment to download (default 0).
Returns:
str: Attachment metadata with either a local file path or download URL.
"""
logger.info(
f"[download_chat_attachment] Message: '{message_id}', Index: {attachment_index}"
)
# Fetch the message to get attachment metadata
msg = await asyncio.to_thread(
service.spaces().messages().get(name=message_id).execute
)
attachments = msg.get("attachment", [])
if not attachments:
return f"No attachments found on message {message_id}."
if attachment_index < 0 or attachment_index >= len(attachments):
return (
f"Invalid attachment_index {attachment_index}. "
f"Message has {len(attachments)} attachment(s) (0-{len(attachments) - 1})."
)
att = attachments[attachment_index]
filename = att.get("contentName", "attachment")
content_type = att.get("contentType", "application/octet-stream")
source = att.get("source", "")
# The media endpoint needs attachmentDataRef.resourceName (e.g.
# "spaces/S/attachments/A"), NOT the attachment name which includes
# the /messages/ segment and causes 400 errors.
media_resource = att.get("attachmentDataRef", {}).get("resourceName", "")
att_name = att.get("name", "")
logger.info(
f"[download_chat_attachment] Downloading '{filename}' ({content_type}), "
f"source={source}, mediaResource={media_resource}, name={att_name}"
)
# Download the attachment binary data via the Chat API media endpoint.
# We use httpx with the Bearer token directly because MediaIoBaseDownload
# and AuthorizedHttp fail in OAuth 2.1 (no refresh_token). The attachment's
# downloadUri points to chat.google.com which requires browser cookies.
if not media_resource and not att_name:
return f"No resource name available for attachment '{filename}'."
# Prefer attachmentDataRef.resourceName for the media endpoint
resource_name = media_resource or att_name
download_url = f"https://chat.googleapis.com/v1/media/{resource_name}?alt=media"
try:
access_token = service._http.credentials.token
async with httpx.AsyncClient(follow_redirects=True) as client:
resp = await client.get(
download_url,
headers={"Authorization": f"Bearer {access_token}"},
)
if resp.status_code != 200:
body = resp.text[:500]
return (
f"Failed to download attachment '{filename}': "
f"HTTP {resp.status_code} from {download_url}\n{body}"
)
file_bytes = resp.content
except Exception as e:
return f"Failed to download attachment '{filename}': {e}"
size_bytes = len(file_bytes)
size_kb = size_bytes / 1024
# Check if we're in stateless mode (can't save files)
from auth.oauth_config import is_stateless_mode
if is_stateless_mode():
b64_preview = base64.urlsafe_b64encode(file_bytes).decode("utf-8")[:100]
return "\n".join(
[
f"Attachment downloaded: {filename} ({content_type})",
f"Size: {size_kb:.1f} KB ({size_bytes} bytes)",
"",
"Stateless mode: File storage disabled.",
f"Base64 preview: {b64_preview}...",
]
)
# Save to local disk
from core.attachment_storage import get_attachment_storage, get_attachment_url
from core.config import get_transport_mode
storage = get_attachment_storage()
b64_data = base64.urlsafe_b64encode(file_bytes).decode("utf-8")
result = storage.save_attachment(
base64_data=b64_data, filename=filename, mime_type=content_type
)
result_lines = [
f"Attachment downloaded: {filename}",
f"Type: {content_type}",
f"Size: {size_kb:.1f} KB ({size_bytes} bytes)",
]
if get_transport_mode() == "stdio":
result_lines.append(f"\nSaved to: {result.path}")
result_lines.append(
"\nThe file has been saved to disk and can be accessed directly via the file path."
)
else:
download_url = get_attachment_url(result.file_id)
result_lines.append(f"\nDownload URL: {download_url}")
result_lines.append("\nThe file will expire after 1 hour.")
logger.info(
f"[download_chat_attachment] Saved {size_kb:.1f} KB attachment to {result.path}"
)
return "\n".join(result_lines)
