# Directory Structure
```
├── Dockerfile
├── google_chat.py
├── LICENSE
├── poetry.lock
├── pyproject.toml
├── README.md
├── requirements.txt
├── server_auth.py
├── server.py
├── src
│ └── mcp_gcp_chat_py
│ └── __init__.py
├── tests
│ └── __init__.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Introduction
This project provides a Google Chat integration for MCP (Model Control Protocol) servers written by Python with FastMCP. It allows you to access and interact with Google Chat spaces and messages through MCP tools.
## Structure
The project consists of two main components:
1. **MCP Server with Google Chat Tools**: Provides tools for interacting with Google Chat through the Model Control Protocol.
- Written by FastMCP
- `server.py`: Main MCP server implementation with Google Chat tools
- `google_chat.py`: Google Chat API integration and authentication handling
2. **Authentication Server**: Standalone component for Google account authentication
- Written by FastAPI
- Handles OAuth2 flow with Google
- Stores and manages access tokens
- Can be run independently or as part of the MCP server
- `server_auth.py`: Authentication server implementation
The authentication flow allows you to obtain and refresh Google API tokens, which are then used by the MCP tools to access Google Chat data. (Your spaces and messages)
## Features
- OAuth2 authentication with Google Chat API
- List available Google Chat spaces
- Retrieve messages from specific spaces with date filtering
- Local authentication server for easy setup
## Requirements
- Python 3.8+
- Google Cloud project with Chat API enabled
- OAuth2 credentials from Google Cloud Console
# How to use?
## Prepare Google Oauth Login
1. Clone this project
```
git clone https://github.com/chy168/google-chat-mcp-server.git
cd google-chat-mcp-server
```
2. Prepare a Google Cloud Project (GCP)
3. Google Cloud Conolse (https://console.cloud.google.com/auth/overview?project=<YOUR_PROJECT_NAME>)
4. Google Auth Platform > Clients > (+) Create client > Web application
reference: https://developers.google.com/identity/protocols/oauth2/?hl=en
Authorized JavaScript origins add: `http://localhost:8000`
Authorized redirect URIs: `http://localhost:8000/auth/callback`
5. After you create a OAuth 2.0 Client, download the client secrets as `.json` file. Save as `credentials.json` at top level of project.
## Run Auth server and get your Google access token (login google only, not MCP server yet)
```
python server.py -local-auth --port 8000
```
- Open browser at http://localhost:8000/auth
- login it!
- after loggined, you access token will be saved as `token.json`
## MCP Configuration (mcp.json)
```
{
"mcpServers": {
"google_chat": {
"command": "uv",
"args": [
"--directory",
"<YOUR_REPO_PATH>/google-chat-mcp-server",
"run",
"server.py",
"--token-path",
"<YOUR_REPO_PATH>/google-chat-mcp-server/token.json"
]
}
}
```
## Tools
The MCP server provides the following tools:
### Google Chat Tools
- `get_chat_spaces()` - List all Google Chat spaces the bot has access to
- `get_space_messages(space_name: str, start_date: str, end_date: str = None)` - List messages from a specific Google Chat space with optional time filtering
## Development and Debug
```
fastmcp dev server.py --with-editable .
```
```
--------------------------------------------------------------------------------
/src/mcp_gcp_chat_py/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/tests/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
fastmcp>=0.1.0
httpx>=0.24.0
google-api-core>=2.11.0
google-api-python-client>=2.0.0
google-auth>=2.22.0
google-auth-httplib2>=0.1.0
google-auth-oauthlib>=1.0.0
fastapi>=0.68.0
uvicorn>=0.15.0
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "mcp-gcp-chat-py"
version = "0.1.0"
description = ""
authors = [
{name = "Zz Chen",email = "[email protected]"}
]
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
"fastmcp (>=0.4.1,<0.5.0)"
]
[tool.poetry]
packages = [{include = "mcp_gcp_chat_py", from = "src"}]
[build-system]
requires = ["poetry-core>=2.0.0,<3.0.0"]
build-backend = "poetry.core.masonry.api"
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
# Use official Python image
FROM python:3.11-slim
# Set work directory
WORKDIR /app
# Install system dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
build-essential \
&& rm -rf /var/lib/apt/lists/*
# Copy project files
COPY . /app
# Install Poetry
RUN pip install poetry
# Install dependencies
RUN poetry config virtualenvs.create false \
&& poetry install --no-interaction --no-ansi --only main
# Expose MCP default port
EXPOSE 8080
# Set environment variables (if needed)
ENV PYTHONUNBUFFERED=1
# Start the MCP server (adjust if entrypoint is different)
CMD ["python", "server.py"]
```
--------------------------------------------------------------------------------
/server.py:
--------------------------------------------------------------------------------
```python
# server.py
import httpx
import sys
import argparse
from typing import List, Dict
from fastmcp import FastMCP
from google_chat import list_chat_spaces, DEFAULT_CALLBACK_URL, set_token_path
from server_auth import run_auth_server
# Create an MCP server
mcp = FastMCP("Demo")
# Add an addition tool
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b
name = "GG"
# Add a dynamic greeting resource
@mcp.resource("greeting://{name}")
def get_greeting(name: str) -> str:
"""Get a personalized greeting"""
return f"Hello, {name}!"
@mcp.tool()
async def fetch_weather(city: str) -> str:
"""Fetch current weather for a city"""
async with httpx.AsyncClient() as client:
response = await client.get(
f"https://api.weather.com/{city}"
)
return response.text
@mcp.tool()
async def get_ip_my_address(city: str) -> str:
"""Get IP address from outian.net"""
async with httpx.AsyncClient() as client:
response = await client.get(
f"http://outian.net/"
)
return response.text
@mcp.tool()
async def get_chat_spaces() -> List[Dict]:
"""List all Google Chat spaces the bot has access to.
This tool requires OAuth authentication. On first run, it will open a browser window
for you to log in with your Google account. Make sure you have credentials.json
downloaded from Google Cloud Console in the current directory.
"""
return await list_chat_spaces()
@mcp.tool()
async def get_space_messages(space_name: str,
start_date: str,
end_date: str = None) -> List[Dict]:
"""List messages from a specific Google Chat space with optional time filtering.
This tool requires OAuth authentication. The space_name should be in the format
'spaces/your_space_id'. Dates should be in YYYY-MM-DD format (e.g., '2024-03-22').
When only start_date is provided, it will query messages for that entire day.
When both dates are provided, it will query messages from start_date 00:00:00Z
to end_date 23:59:59Z.
Args:
space_name: The name/identifier of the space to fetch messages from
start_date: Required start date in YYYY-MM-DD format
end_date: Optional end date in YYYY-MM-DD format
Returns:
List of message objects from the space matching the time criteria
Raises:
ValueError: If the date format is invalid or dates are in wrong order
"""
from google_chat import list_space_messages
from datetime import datetime, timezone
try:
# Parse start date and set to beginning of day (00:00:00Z)
start_datetime = datetime.strptime(start_date, '%Y-%m-%d').replace(
hour=0, minute=0, second=0, microsecond=0, tzinfo=timezone.utc
)
# Parse end date if provided and set to end of day (23:59:59Z)
end_datetime = None
if end_date:
end_datetime = datetime.strptime(end_date, '%Y-%m-%d').replace(
hour=23, minute=59, second=59, microsecond=999999, tzinfo=timezone.utc
)
# Validate date range
if start_datetime > end_datetime:
raise ValueError("start_date must be before end_date")
except ValueError as e:
if "strptime" in str(e):
raise ValueError("Dates must be in YYYY-MM-DD format (e.g., '2024-03-22')")
raise e
return await list_space_messages(space_name, start_datetime, end_datetime)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description='MCP Server with Google Chat Authentication')
parser.add_argument('-local-auth', action='store_true', help='Run the local authentication server')
parser.add_argument('--host', default='localhost', help='Host to bind the server to (default: localhost)')
parser.add_argument('--port', type=int, default=8000, help='Port to run the server on (default: 8000)')
parser.add_argument('--token-path', default='token.json', help='Path to store OAuth token (default: token.json)')
args = parser.parse_args()
# Set the token path for OAuth storage
set_token_path(args.token_path)
if args.local_auth:
print(f"\nStarting local authentication server at http://{args.host}:{args.port}")
print("Available endpoints:")
print(" - /auth : Start OAuth authentication flow")
print(" - /status : Check authentication status")
print(" - /auth/callback : OAuth callback endpoint")
print(f"\nDefault callback URL: {DEFAULT_CALLBACK_URL}")
print(f"Token will be stored at: {args.token_path}")
print("\nPress CTRL+C to stop the server")
print("-" * 50)
run_auth_server(port=args.port, host=args.host)
else:
mcp.run()
```
--------------------------------------------------------------------------------
/google_chat.py:
--------------------------------------------------------------------------------
```python
import os
import json
import datetime
from typing import List, Dict, Optional, Tuple
from google.oauth2.credentials import Credentials
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from googleapiclient.discovery import build
from pathlib import Path
# If modifying these scopes, delete the file token.json.
SCOPES = [
'https://www.googleapis.com/auth/chat.spaces.readonly',
'https://www.googleapis.com/auth/chat.messages'
]
DEFAULT_CALLBACK_URL = "http://localhost:8000/auth/callback"
DEFAULT_TOKEN_PATH = 'token.json'
# Store credentials info
token_info = {
'credentials': None,
'last_refresh': None,
'token_path': DEFAULT_TOKEN_PATH
}
def set_token_path(path: str) -> None:
"""Set the global token path for OAuth storage.
Args:
path: Path where the token should be stored
"""
token_info['token_path'] = path
def save_credentials(creds: Credentials, token_path: Optional[str] = None) -> None:
"""Save credentials to file and update in-memory cache.
Args:
creds: The credentials to save
token_path: Path to save the token file
"""
# Use configured token path if none provided
if token_path is None:
token_path = token_info['token_path']
# Save to file
token_path = Path(token_path)
with open(token_path, 'w') as token:
token.write(creds.to_json())
# Update in-memory cache
token_info['credentials'] = creds
token_info['last_refresh'] = datetime.datetime.utcnow()
def get_credentials(token_path: Optional[str] = None) -> Optional[Credentials]:
"""Gets valid user credentials from storage or memory.
Args:
token_path: Optional path to token file. If None, uses the configured path.
Returns:
Credentials object or None if no valid credentials exist
"""
if token_path is None:
token_path = token_info['token_path']
creds = token_info['credentials']
# If no credentials in memory, try to load from file
if not creds:
token_path = Path(token_path)
if token_path.exists():
creds = Credentials.from_authorized_user_file(str(token_path), SCOPES)
token_info['credentials'] = creds
# If we have credentials that need refresh
if creds and creds.expired and creds.refresh_token:
try:
creds.refresh(Request())
save_credentials(creds, token_path)
except Exception:
return None
return creds if (creds and creds.valid) else None
async def refresh_token(token_path: Optional[str] = None) -> Tuple[bool, str]:
"""Attempt to refresh the current token.
Args:
token_path: Path to the token file. If None, uses the configured path.
Returns:
Tuple of (success: bool, message: str)
"""
if token_path is None:
token_path = token_info['token_path']
try:
creds = token_info['credentials']
if not creds:
token_path = Path(token_path)
if not token_path.exists():
return False, "No token file found"
creds = Credentials.from_authorized_user_file(str(token_path), SCOPES)
if not creds.refresh_token:
return False, "No refresh token available"
creds.refresh(Request())
save_credentials(creds, token_path)
return True, "Token refreshed successfully"
except Exception as e:
return False, f"Failed to refresh token: {str(e)}"
# MCP functions
async def list_chat_spaces() -> List[Dict]:
"""Lists all Google Chat spaces the bot has access to."""
try:
creds = get_credentials()
if not creds:
raise Exception("No valid credentials found. Please authenticate first.")
service = build('chat', 'v1', credentials=creds)
spaces = service.spaces().list(pageSize=30).execute()
return spaces.get('spaces', [])
except Exception as e:
raise Exception(f"Failed to list chat spaces: {str(e)}")
async def list_space_messages(space_name: str,
start_date: Optional[datetime.datetime] = None,
end_date: Optional[datetime.datetime] = None) -> List[Dict]:
"""Lists messages from a specific Google Chat space with optional time filtering.
Args:
space_name: The name/identifier of the space to fetch messages from
start_date: Optional start datetime for filtering messages. If provided without end_date,
will query messages for the entire day of start_date
end_date: Optional end datetime for filtering messages. Only used if start_date is also provided
Returns:
List of message objects from the space matching the time criteria
Raises:
Exception: If authentication fails or API request fails
"""
try:
creds = get_credentials()
if not creds:
raise Exception("No valid credentials found. Please authenticate first.")
service = build('chat', 'v1', credentials=creds)
# Prepare filter string based on provided dates
filter_str = None
if start_date:
if end_date:
# Format for date range query
filter_str = f"createTime > \"{start_date.isoformat()}\" AND createTime < \"{end_date.isoformat()}\""
else:
# For single day query, set range from start of day to end of day
day_start = start_date.replace(hour=0, minute=0, second=0, microsecond=0)
day_end = day_start + datetime.timedelta(days=1)
filter_str = f"createTime > \"{day_start.isoformat()}\" AND createTime < \"{day_end.isoformat()}\""
# Make API request
request = service.spaces().messages().list(parent=space_name, pageSize=100)
if filter_str:
request = service.spaces().messages().list(parent=space_name, filter=filter_str, pageSize=100)
response = request.execute()
return response.get('messages', [])
except Exception as e:
raise Exception(f"Failed to list messages in space: {str(e)}")
```
--------------------------------------------------------------------------------
/server_auth.py:
--------------------------------------------------------------------------------
```python
from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import JSONResponse, RedirectResponse
import uvicorn
import signal
import asyncio
from pathlib import Path
from typing import Optional, Dict
from datetime import datetime
from google_auth_oauthlib.flow import InstalledAppFlow
from google_chat import (
get_credentials,
save_credentials,
refresh_token,
SCOPES,
DEFAULT_CALLBACK_URL,
token_info
)
# Store OAuth flow state
oauth_flows: Dict[str, InstalledAppFlow] = {}
# Create FastAPI app for local auth server
app = FastAPI(title="Google Chat Auth Server")
@app.get("/auth")
async def start_auth(callback_url: Optional[str] = Query(None)):
"""Start OAuth authentication flow"""
try:
# Check if we already have valid credentials
if get_credentials():
return JSONResponse(
content={
"status": "already_authenticated",
"message": "Valid credentials already exist"
}
)
# Initialize OAuth 2.0 flow
credentials_path = Path('credentials.json')
if not credentials_path.exists():
raise FileNotFoundError(
"credentials.json not found. Please download it from Google Cloud Console "
"and save it in the current directory."
)
flow = InstalledAppFlow.from_client_secrets_file(
str(credentials_path),
SCOPES,
redirect_uri=callback_url or DEFAULT_CALLBACK_URL
)
# Generate authorization URL with offline access and force approval
auth_url, state = flow.authorization_url(
access_type='offline', # Enable offline access
prompt='consent', # Force consent screen to ensure refresh token
include_granted_scopes='true'
)
# Store the flow object for later use
oauth_flows[state] = flow
# Redirect user to Google's auth page
return RedirectResponse(url=auth_url)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/auth/callback")
async def auth_callback(
state: str = Query(...),
code: Optional[str] = Query(None),
error: Optional[str] = Query(None)
):
"""Handle OAuth callback"""
try:
if error:
raise HTTPException(
status_code=400,
detail=f"Authorization failed: {error}"
)
if not code:
raise HTTPException(
status_code=400,
detail="No authorization code received"
)
# Retrieve the flow object
flow = oauth_flows.get(state)
if not flow:
raise HTTPException(
status_code=400,
detail="Invalid state parameter"
)
try:
# Exchange auth code for credentials with offline access
print("fetching token: ", code)
flow.fetch_token(
code=code,
# Ensure we're requesting offline access for refresh tokens
access_type='offline'
)
print("fetched credentials: ", flow.credentials)
creds = flow.credentials
# Verify we got a refresh token
if not creds.refresh_token:
raise HTTPException(
status_code=400,
detail="Failed to obtain refresh token. Please try again."
)
# Save credentials both to file and memory
print("saving credentials: ", creds)
save_credentials(creds)
# Clean up the flow object
del oauth_flows[state]
return JSONResponse(
content={
"status": "success",
"message": "Authorization successful. Long-lived token obtained. You can close this window.",
"token_file": token_info['token_path'],
"expires_at": creds.expiry.isoformat() if creds.expiry else None,
"has_refresh_token": bool(creds.refresh_token)
}
)
except Exception as e:
# Clean up flow object even if there's an error
del oauth_flows[state]
raise
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/auth/refresh")
async def manual_token_refresh():
"""Manually trigger a token refresh"""
success, message = await refresh_token()
if success:
creds = token_info['credentials']
return JSONResponse(
content={
"status": "success",
"message": message,
"expires_at": creds.expiry.isoformat() if creds.expiry else None,
"last_refresh": token_info['last_refresh'].isoformat()
}
)
else:
raise HTTPException(
status_code=400,
detail=message
)
@app.get("/status")
async def check_auth_status():
"""Check if we have valid credentials"""
token_path = token_info['token_path']
token_file = Path(token_path)
if not token_file.exists():
return JSONResponse(
content={
"status": "not_authenticated",
"message": "No authentication token found",
"token_path": str(token_path)
}
)
try:
creds = get_credentials()
if creds:
return JSONResponse(
content={
"status": "authenticated",
"message": "Valid credentials exist",
"token_path": str(token_path),
"expires_at": creds.expiry.isoformat() if creds.expiry else None,
"last_refresh": token_info['last_refresh'].isoformat() if token_info['last_refresh'] else None,
"has_refresh_token": bool(creds.refresh_token)
}
)
else:
return JSONResponse(
content={
"status": "expired",
"message": "Credentials exist but are expired or invalid",
"token_path": str(token_path)
}
)
except Exception as e:
return JSONResponse(
content={
"status": "error",
"message": str(e),
"token_path": str(token_path)
},
status_code=500
)
def run_auth_server(port: int = 8000, host: str = "localhost"):
"""Run the authentication server with graceful shutdown support
Args:
port: Port to run the server on (default: 8000)
host: Host to bind the server to (default: localhost)
"""
server_config = uvicorn.Config(app, host=host, port=port)
server = uvicorn.Server(server_config)
# Handle graceful shutdown
def signal_handler(signum, frame):
print("\nReceived signal to terminate. Performing graceful shutdown...")
asyncio.create_task(server.shutdown())
# Register signal handlers
signal.signal(signal.SIGINT, signal_handler) # Handle Ctrl+C
signal.signal(signal.SIGTERM, signal_handler) # Handle termination signal
try:
print(f"\nServer is running at: http://{host}:{port}")
print(f"Default callback URL: {DEFAULT_CALLBACK_URL}")
# Start the server
server.run()
except KeyboardInterrupt:
print("\nShutting down the auth server...")
finally:
print("Auth server has been stopped.")
```