ayush3298 · March 9, 2025 18:28
diff --git a/gdrive.py b/gdrive.py
 #!/usr/bin/env python3
 """
 Google Drive MCP Server

 This server provides Model Context Protocol (MCP) integration with Google Drive,
 allowing clients to:
 - List and search files in Google Drive
 - Read file contents
 - List, read, and write Google Sheet data
 """

 import argparse
 import asyncio
 import json
 import os
 import sys
 from functools import wraps
 from typing import Any, Dict, List, Optional, Union

 import aiohttp
 from google.auth.transport.requests import Request
 from google.oauth2.credentials import Credentials
 from google_auth_oauthlib.flow import InstalledAppFlow
 from googleapiclient.discovery import build
 from googleapiclient.errors import HttpError
 from googleapiclient.http import MediaIoBaseDownload
 from mcp.server.fastmcp import FastMCP
 import pandas as pd
 import io

 # If modifying these scopes, delete the file token.json.
 SCOPES = [
    'https://www.googleapis.com/auth/drive',
    'https://www.googleapis.com/auth/spreadsheets',
 ]

 # Initialize FastMCP server
 mcp = FastMCP("google-drive")

 # Global variable to store the authenticated service
 drive_service = None
 sheets_service = None


 def require_auth(func):
    """Decorator to ensure Google API services are authenticated."""
    @wraps(func)
    async def wrapper(*args, **kwargs):
        global drive_service, sheets_service
        if drive_service is None or sheets_service is None:
            await authenticate()
        return await func(*args, **kwargs)
    return wrapper


 async def authenticate():
    """Authenticate with Google Drive and Sheets APIs."""
    global drive_service, sheets_service
    
    creds = None
    # The file token.json stores the user's access and refresh tokens
    if os.path.exists('token.json'):
        creds = Credentials.from_authorized_user_info(
            json.loads(open('token.json').read()), SCOPES)
    
    # If there are no (valid) credentials available, let the user log in.
    if not creds or not creds.valid:
        if creds and creds.expired and creds.refresh_token:
            creds.refresh(Request())
        else:
            if not os.path.exists('credentials.json'):
                raise FileNotFoundError(
                    "credentials.json file not found. Please download it from "
                    "the Google Cloud Console and place it in the current directory."
                )
            flow = InstalledAppFlow.from_client_secrets_file(
                'credentials.json', SCOPES)
            creds = flow.run_local_server(port=0)
        
        # Save the credentials for the next run
        with open('token.json', 'w') as token:
            token.write(creds.to_json())
    
    drive_service = build('drive', 'v3', credentials=creds)
    sheets_service = build('sheets', 'v4', credentials=creds)
    
    return True


 # Resources
 @mcp.resource(
    uri_schema="gdrive://list",
    name="Google Drive File List",
    description="List files in Google Drive"
 )
 @require_auth
 async def list_files() -> str:
    """Lists files from Google Drive."""
    try:
        results = drive_service.files().list(
            pageSize=50,
            fields="nextPageToken, files(id, name, mimeType, webViewLink)"
        ).execute()
        items = results.get('files', [])
        
        if not items:
            return "No files found in Google Drive."
        
        file_list = []
        for item in items:
            file_list.append({
                'id': item['id'],
                'name': item['name'],
                'mimeType': item['mimeType'],
                'webViewLink': item.get('webViewLink', 'N/A')
            })
        
        return json.dumps(file_list, indent=2)
    
    except HttpError as error:
        return f"An error occurred listing files: {error}"


 @mcp.resource(
    uri_schema="gdrive://file/{file_id}",
    name="Google Drive File Content",
    description="Get the content of a file from Google Drive by ID"
 )
 @require_auth
 async def get_file_content(file_id: str) -> str:
    """Get the content of a file by ID."""
    try:
        # Get the file metadata to determine the type
        file_metadata = drive_service.files().get(fileId=file_id, fields="name,mimeType").execute()
        mime_type = file_metadata['mimeType']
        file_name = file_metadata['name']
        
        # Handle different file types
        if mime_type == 'application/vnd.google-apps.spreadsheet':
            # For Google Sheets, get the data using Sheets API
            result = sheets_service.spreadsheets().values().get(
                spreadsheetId=file_id, range='A1:Z1000').execute()
            rows = result.get('values', [])
            
            if not rows:
                return f"No data found in the sheet: {file_name}"
            
            # Format as CSV
            csv_content = io.StringIO()
            for row in rows:
                csv_content.write(','.join([str(cell) for cell in row]) + '\n')
            
            return csv_content.getvalue()
        
        elif mime_type in ['text/plain', 'text/csv', 'application/json']:
            # For text files, download the content
            request = drive_service.files().get_media(fileId=file_id)
            file_content = io.BytesIO()
            downloader = MediaIoBaseDownload(file_content, request)
            
            done = False
            while not done:
                status, done = downloader.next_chunk()
            
            return file_content.getvalue().decode('utf-8')
        
        else:
            return f"File '{file_name}' has unsupported MIME type: {mime_type}. Only text, CSV, JSON, and Google Sheets are supported."
    
    except HttpError as error:
        return f"An error occurred retrieving file: {error}"


 @mcp.resource(
    uri_schema="gdrive://sheets/list",
    name="Google Sheets List",
    description="List all Google Sheets"
 )
 @require_auth
 async def list_sheets() -> str:
    """Lists all Google Sheets from Drive."""
    try:
        results = drive_service.files().list(
            q="mimeType='application/vnd.google-apps.spreadsheet'",
            pageSize=50,
            fields="nextPageToken, files(id, name, webViewLink)"
        ).execute()
        items = results.get('files', [])
        
        if not items:
            return "No Google Sheets found."
        
        sheets_list = []
        for item in items:
            sheets_list.append({
                'id': item['id'],
                'name': item['name'],
                'webViewLink': item.get('webViewLink', 'N/A')
            })
        
        return json.dumps(sheets_list, indent=2)
    
    except HttpError as error:
        return f"An error occurred listing sheets: {error}"


 # Tools
 @mcp.tool()
 @require_auth
 async def search_files(query: str) -> str:
    """Search for files in Google Drive.
    
    Args:
        query: Search query text (e.g., "budget 2024" or "type:spreadsheet")
    """
    try:
        results = drive_service.files().list(
            q=f"fullText contains '{query}'",
            pageSize=20,
            fields="nextPageToken, files(id, name, mimeType, webViewLink)"
        ).execute()
        items = results.get('files', [])
        
        if not items:
            return f"No files found matching query: {query}"
        
        file_list = []
        for item in items:
            file_list.append({
                'id': item['id'],
                'name': item['name'],
                'mimeType': item['mimeType'],
                'webViewLink': item.get('webViewLink', 'N/A')
            })
        
        return json.dumps(file_list, indent=2)
    
    except HttpError as error:
        return f"An error occurred searching files: {error}"


 @mcp.tool()
 @require_auth
 async def read_sheet_range(spreadsheet_id: str, range_name: str = "A1:Z1000") -> str:
    """Read data from a specific range in a Google Sheet.
    
    Args:
        spreadsheet_id: The ID of the Google Sheet
        range_name: The A1 notation of the range to read (default: A1:Z1000)
    """
    try:
        result = sheets_service.spreadsheets().values().get(
            spreadsheetId=spreadsheet_id, range=range_name).execute()
        rows = result.get('values', [])
        
        if not rows:
            return f"No data found in range: {range_name}"
        
        # Try to convert to a cleaner tabular format
        try:
            df = pd.DataFrame(rows)
            # Use first row as headers if there are at least 2 rows
            if len(rows) > 1:
                df.columns = df.iloc[0]
                df = df.drop(0)
            
            return df.to_string(index=False)
        except Exception:
            # Fall back to simple formatting if pandas conversion fails
            formatted_rows = []
            for row in rows:
                formatted_rows.append('\t'.join([str(cell) for cell in row]))
            
            return '\n'.join(formatted_rows)
    
    except HttpError as error:
        return f"An error occurred reading sheet range: {error}"


 @mcp.tool()
 @require_auth
 async def write_to_sheet(spreadsheet_id: str, range_name: str, data: str) -> str:
    """Write data to a Google Sheet.
    
    Args:
        spreadsheet_id: The ID of the Google Sheet
        range_name: The A1 notation of the range to write (e.g., "Sheet1!A1:B5")
        data: CSV-formatted data to write (each line is a row, values separated by commas)
    """
    try:
        # Convert the CSV-formatted data to a list of lists
        lines = data.strip().split('\n')
        values = [line.split(',') for line in lines]
        
        body = {
            'values': values
        }
        
        result = sheets_service.spreadsheets().values().update(
            spreadsheetId=spreadsheet_id, 
            range=range_name,
            valueInputOption='USER_ENTERED',
            body=body
        ).execute()
        
        return f"Updated {result.get('updatedCells')} cells in {range_name}"
    
    except HttpError as error:
        return f"An error occurred writing to sheet: {error}"


 @mcp.tool()
 @require_auth
 async def create_new_sheet(title: str) -> str:
    """Create a new Google Sheet.
    
    Args:
        title: Title for the new spreadsheet
    """
    try:
        spreadsheet_body = {
            'properties': {
                'title': title
            }
        }
        
        spreadsheet = sheets_service.spreadsheets().create(
            body=spreadsheet_body).execute()
        
        # Get the spreadsheet ID and URL
        spreadsheet_id = spreadsheet.get('spreadsheetId')
        spreadsheet_url = spreadsheet.get('spreadsheetUrl')
        
        return json.dumps({
            "message": f"Created new spreadsheet: {title}",
            "id": spreadsheet_id,
            "url": spreadsheet_url
        }, indent=2)
    
    except HttpError as error:
        return f"An error occurred creating the spreadsheet: {error}"


 @mcp.tool()
 @require_auth
 async def run_sheet_query(spreadsheet_id: str, sheet_name: str, query: str) -> str:
    """Run a simplified SQL-like query on a Google Sheet.
    
    Args:
        spreadsheet_id: The ID of the Google Sheet
        sheet_name: The name of the sheet to query
        query: SQL-like query (supports: SELECT, WHERE, GROUP BY, ORDER BY, LIMIT)
    """
    try:
        # Get the sheet data first
        result = sheets_service.spreadsheets().values().get(
            spreadsheetId=spreadsheet_id, range=f"{sheet_name}").execute()
        rows = result.get('values', [])
        
        if not rows:
            return f"No data found in sheet: {sheet_name}"
        
        # Convert to pandas DataFrame
        df = pd.DataFrame(rows[1:], columns=rows[0])
        
        # Interpret the query
        if not query.lower().startswith('select'):
            query = 'SELECT * ' + query
        
        # Handle basic SQL operations
        # This is a very simplified SQL parser for demo purposes
        query_parts = query.lower().split()
        
        # Extract where clause if it exists
        where_clause = None
        if 'where' in query_parts:
            where_index = query_parts.index('where')
            # Find the next clause after WHERE
            next_clause_keywords = ['group', 'order', 'limit']
            next_clause_indices = [query_parts.index(kw) for kw in next_clause_keywords if kw in query_parts]
            
            if next_clause_indices:
                end_index = min(next_clause_indices)
                where_clause = ' '.join(query_parts[where_index+1:end_index])
            else:
                where_clause = ' '.join(query_parts[where_index+1:])
        
        # Extract order by clause
        order_by = None
        if 'order' in query_parts and 'by' in query_parts:
            order_index = query_parts.index('order')
            if query_parts[order_index + 1] == 'by':
                # Same logic to find the end of the ORDER BY clause
                next_clause_keywords = ['limit']
                next_clause_indices = [query_parts.index(kw) for kw in next_clause_keywords if kw in query_parts]
                
                if next_clause_indices:
                    end_index = min(next_clause_indices)
                    order_by = ' '.join(query_parts[order_index+2:end_index])
                else:
                    order_by = ' '.join(query_parts[order_index+2:])
        
        # Extract limit
        limit = None
        if 'limit' in query_parts:
            limit_index = query_parts.index('limit')
            if limit_index + 1 < len(query_parts):
                try:
                    limit = int(query_parts[limit_index + 1])
                except ValueError:
                    pass
        
        # Apply filters through Python code
        # Note: This is a very basic implementation
        if where_clause:
            # Here we'd need a more sophisticated parser for real SQL WHERE clauses
            # This is just a simple example that handles basic conditions
            try:
                # For simple conditions like "column = value"
                parts = where_clause.split('=')
                if len(parts) == 2:
                    col = parts[0].strip()
                    val = parts[1].strip().strip("'\"")
                    df = df[df[col] == val]
            except Exception as e:
                return f"Error parsing WHERE clause: {e}"
        
        # Apply ordering
        if order_by:
            try:
                # Simple order by implementation
                parts = order_by.split()
                col = parts[0].strip()
                ascending = True
                if len(parts) > 1 and parts[1].upper() == 'DESC':
                    ascending = False
                df = df.sort_values(by=col, ascending=ascending)
            except Exception as e:
                return f"Error applying ORDER BY: {e}"
        
        # Apply limit
        if limit:
            df = df.head(limit)
        
        # Return the result
        return df.to_string(index=False)
    
    except HttpError as error:
        return f"An error occurred querying the sheet: {error}"


 @mcp.tool()
 @require_auth
 async def get_file_metadata(file_id: str) -> str:
    """Get detailed metadata for a specific file by ID.
    
    Args:
        file_id: The ID of the file in Google Drive
    """
    try:
        file = drive_service.files().get(
            fileId=file_id,
            fields="id, name, mimeType, description, createdTime, modifiedTime, webViewLink, size"
        ).execute()
        
        return json.dumps(file, indent=2)
    
    except HttpError as error:
        return f"An error occurred getting file metadata: {error}"


 # Prompts
 @mcp.prompt()
 @require_auth
 async def analyze_sheet(sheet_id: str, analysis_type: str = "summary") -> List[Dict[str, Any]]:
    """Prompt for analyzing Google Sheet data.
    
    Args:
        sheet_id: ID of the Google Sheet to analyze
        analysis_type: Type of analysis (summary, trends, outliers)
    """
    result = sheets_service.spreadsheets().values().get(
        spreadsheetId=sheet_id, range="A1:Z1000").execute()
    rows = result.get('values', [])
    
    if not rows:
        return [
            {
                "role": "user",
                "content": f"The Google Sheet with ID '{sheet_id}' is empty. Please provide data before analysis."
            }
        ]
    
    # Get the metadata for the sheet name
    sheet_metadata = drive_service.files().get(fileId=sheet_id, fields="name").execute()
    sheet_name = sheet_metadata.get('name', 'Unnamed Sheet')
    
    # Create a DataFrame for analysis
    df = pd.DataFrame(rows[1:], columns=rows[0])
    
    # Generate different prompts based on analysis type
    if analysis_type.lower() == "summary":
        return [
            {
                "role": "user",
                "content": f"""
                Analyze this Google Sheet data from '{sheet_name}':
                
                Column Headers: {', '.join(df.columns.tolist())}
                
                Number of Rows: {len(df)}
                
                Sample Data (first 5 rows):
                {df.head(5).to_string(index=False)}
                
                Please provide:
                1. A summary of what this data represents
                2. Key statistics for numeric columns
                3. Distribution of categorical data
                4. Any notable patterns
                5. Recommendations for further analysis
                """
            }
        ]
    
    elif analysis_type.lower() == "trends":
        return [
            {
                "role": "user",
                "content": f"""
                Analyze trends in this Google Sheet data from '{sheet_name}':
                
                Column Headers: {', '.join(df.columns.tolist())}
                
                Sample Data (first 5 rows):
                {df.head(5).to_string(index=False)}
                
                Please identify:
                1. Temporal trends (if date/time columns exist)
                2. Correlation between numeric columns
                3. Changes in distributions over any sequential variables
                4. Growth rates or decline patterns
                5. Cyclical patterns if they exist
                6. Recommendations for visualizing these trends
                """
            }
        ]
    
    elif analysis_type.lower() == "outliers":
        return [
            {
                "role": "user",
                "content": f"""
                Analyze outliers in this Google Sheet data from '{sheet_name}':
                
                Column Headers: {', '.join(df.columns.tolist())}
                
                Please identify:
                1. Statistical outliers in numeric columns
                2. Unusual patterns or combinations in the data
                3. Potential data entry errors
                4. Impact of outliers on overall analysis
                5. Recommendations for handling these outliers
                
                Here's a statistical summary of numeric columns:
                {df.describe().to_string()}
                """
            }
        ]
    
    else:
        return [
            {
                "role": "user",
                "content": f"""
                Analyze this Google Sheet data from '{sheet_name}' based on your judgment:
                
                Column Headers: {', '.join(df.columns.tolist())}
                
                Sample Data (first 10 rows):
                {df.head(10).to_string(index=False)}
                
                Please provide a comprehensive analysis including key insights and recommendations.
                """
            }
        ]


 @mcp.prompt()
 @require_auth
 async def create_report_template(report_type: str) -> List[Dict[str, Any]]:
    """Prompt for creating a Google Sheets report template.
    
    Args:
        report_type: Type of report (financial, project_tracker, dashboard, inventory)
    """
    report_prompts = {
        "financial": """
            Create a comprehensive financial report template in Google Sheets with the following components:
            
            1. Income Statement sheet with:
               - Revenue categories
               - Expense categories
               - Profit calculations
               - Month-to-month comparisons
               - Year-to-date totals
            
            2. Balance Sheet with:
               - Assets section
               - Liabilities section
               - Equity section
               - Auto-calculated totals
            
            3. Cash Flow Statement with:
               - Operating activities
               - Investing activities
               - Financing activities
               
            4. Financial Ratios dashboard with:
               - Liquidity ratios
               - Profitability ratios
               - Leverage ratios
               - Efficiency ratios
               
            Include formulas, conditional formatting, and data validation where appropriate.
            Explain the key components and how to use them effectively.
        """,
        
        "project_tracker": """
            Design a comprehensive project management tracker in Google Sheets with the following components:
            
            1. Project Overview dashboard with:
               - Project name, description, objectives
               - Key dates and milestones
               - Budget summary
               - Overall status indicators
            
            2. Task Tracker sheet with:
               - Task descriptions
               - Assignees
               - Due dates
               - Status (Not Started, In Progress, Completed, Delayed)
               - Priority levels
               - Dependencies
               - Progress percentage
               - Notes/comments
            
            3. Resource Allocation sheet with:
               - Team members
               - Hours allocation
               - Skills matrix
               - Availability calendar
            
            4. Budget Tracker with:
               - Planned vs actual expenses
               - Cost categories
               - Variance calculations
               
            5. Risk Register with:
               - Risk descriptions
               - Impact and probability ratings
               - Mitigation strategies
               - Status tracking
               
            Include formulas for automatic calculations, conditional formatting for visual cues, 
            and data validation to maintain data integrity. Explain how each component works and
            suggest best practices for using this tracker effectively.
        """,
        
        "dashboard": """
            Create a versatile business KPI dashboard template in Google Sheets with the following components:
            
            1. Executive Summary with:
               - Key performance indicators at a glance
               - Performance vs targets visualization
               - Period-over-period comparisons
               - Traffic light indicators for status
            
            2. Sales Performance section with:
               - Revenue by product/service
               - Sales by region/channel
               - Conversion metrics
               - Sales pipeline visualization
            
            3. Marketing Metrics section with:
               - Campaign performance
               - Customer acquisition cost
               - Channel effectiveness
               - Engagement metrics
            
            4. Customer Insights section with:
               - Customer satisfaction scores
               - Retention metrics
               - Lifetime value calculations
               - Churn analysis
            
            5. Operations section with:
               - Production/service delivery metrics
               - Quality indicators
               - Efficiency measures
               - Inventory levels (if applicable)
            
            6. Financial Overview with:
               - Profit margins
               - Cash flow status
               - Expense breakdown
               - Financial forecasts
            
            Include instructions for data input, formulas for calculations, and guidance on creating
            visualizations like charts and sparklines. Design the dashboard for clarity and ease of interpretation
            with conditional formatting and appropriate data presentation techniques.
        """,
        
        "inventory": """
            Design a comprehensive inventory management system in Google Sheets with the following components:
            
            1. Inventory Dashboard with:
               - Total inventory value
               - Stock level summary
               - Low stock alerts
               - Inventory turnover metrics
               - Value by category/location
            
            2. Master Inventory List with:
               - Item IDs and descriptions
               - Categories and subcategories
               - Quantities on hand
               - Reorder points
               - Cost and retail pricing
               - Supplier information
               - Storage location
               - Last count date
            
            3. Inventory Transactions Log with:
               - Date and time of transactions
               - Transaction types (receiving, sales, adjustments, transfers)
               - Quantity changes
               - User tracking
               - Notes/reasons
            
            4. Supplier Management section with:
               - Supplier details
               - Order history
               - Performance metrics
               - Contact information
            
            5. Reports and Analytics section with:
               - Sales velocity by product
               - Dead stock identification
               - Seasonal trends
               - Category performance
               - Forecasting tools
            
            Include formulas for automatic calculations (like FIFO/LIFO valuation), data validation for entry,
            conditional formatting for visual alerts, and instructions on implementing barcode scanning
            if applicable. Explain how each component works together and best practices for maintaining
            accurate inventory records.
        """
    }
    
    # Default generic report if type not found
    default_prompt = """
        Create a professional report template in Google Sheets with:
        
        1. An executive summary dashboard
        2. Detailed data tables
        3. Analysis sections
        4. Visualizations and charts
        5. Conclusions and recommendations
        
        Include appropriate formatting, formulas, and instructions for users to customize for their needs.
    """
    
    # Get the specific prompt or fallback to default
    prompt_content = report_prompts.get(report_type.lower(), default_prompt)
    
    return [
        {
            "role": "user",
            "content": prompt_content.strip()
        }
    ]


 def parse_args():
    """Parse command line arguments."""
    parser = argparse.ArgumentParser(description='Google Drive MCP Server')
    parser.add_argument('--stdio', action='store_true', help='Use stdio transport')
    return parser.parse_args()


 if __name__ == "__main__":
    args = parse_args()
    
    # Start the server with specified transport
    if args.stdio:
        mcp.run(transport='stdio')
    else:
        # Default to stdio if no transport specified
        mcp.run(transport='stdio')
	#!/usr/bin/env python3
	"""
	Google Drive MCP Server

	This server provides Model Context Protocol (MCP) integration with Google Drive,
	allowing clients to:
	- List and search files in Google Drive
	- Read file contents
	- List, read, and write Google Sheet data
	"""

	import argparse
	import asyncio
	import json
	import os
	import sys
	from functools import wraps
	from typing import Any, Dict, List, Optional, Union

	import aiohttp
	from google.auth.transport.requests import Request
	from google.oauth2.credentials import Credentials
	from google_auth_oauthlib.flow import InstalledAppFlow
	from googleapiclient.discovery import build
	from googleapiclient.errors import HttpError
	from googleapiclient.http import MediaIoBaseDownload
	from mcp.server.fastmcp import FastMCP
	import pandas as pd
	import io

	# If modifying these scopes, delete the file token.json.
	SCOPES = [
	'https://www.googleapis.com/auth/drive',
	'https://www.googleapis.com/auth/spreadsheets',
	]

	# Initialize FastMCP server
	mcp = FastMCP("google-drive")

	# Global variable to store the authenticated service
	drive_service = None
	sheets_service = None


	def require_auth(func):
	"""Decorator to ensure Google API services are authenticated."""
	@wraps(func)
	async def wrapper(args, *kwargs):
	global drive_service, sheets_service
	if drive_service is None or sheets_service is None:
	await authenticate()
	return await func(args, *kwargs)
	return wrapper


	async def authenticate():
	"""Authenticate with Google Drive and Sheets APIs."""
	global drive_service, sheets_service

	creds = None
	# The file token.json stores the user's access and refresh tokens
	if os.path.exists('token.json'):
	creds = Credentials.from_authorized_user_info(
	json.loads(open('token.json').read()), SCOPES)

	# If there are no (valid) credentials available, let the user log in.
	if not creds or not creds.valid:
	if creds and creds.expired and creds.refresh_token:
	creds.refresh(Request())
	else:
	if not os.path.exists('credentials.json'):
	raise FileNotFoundError(
	"credentials.json file not found. Please download it from "
	"the Google Cloud Console and place it in the current directory."
	)
	flow = InstalledAppFlow.from_client_secrets_file(
	'credentials.json', SCOPES)
	creds = flow.run_local_server(port=0)

	# Save the credentials for the next run
	with open('token.json', 'w') as token:
	token.write(creds.to_json())

	drive_service = build('drive', 'v3', credentials=creds)
	sheets_service = build('sheets', 'v4', credentials=creds)

	return True


	# Resources
	@mcp.resource(
	uri_schema="gdrive://list",
	name="Google Drive File List",
	description="List files in Google Drive"
	)
	@require_auth
	async def list_files() -> str:
	"""Lists files from Google Drive."""
	try:
	results = drive_service.files().list(
	pageSize=50,
	fields="nextPageToken, files(id, name, mimeType, webViewLink)"
	).execute()
	items = results.get('files', [])

	if not items:
	return "No files found in Google Drive."

	file_list = []
	for item in items:
	file_list.append({
	'id': item['id'],
	'name': item['name'],
	'mimeType': item['mimeType'],
	'webViewLink': item.get('webViewLink', 'N/A')
	})

	return json.dumps(file_list, indent=2)

	except HttpError as error:
	return f"An error occurred listing files: {error}"


	@mcp.resource(
	uri_schema="gdrive://file/{file_id}",
	name="Google Drive File Content",
	description="Get the content of a file from Google Drive by ID"
	)
	@require_auth
	async def get_file_content(file_id: str) -> str:
	"""Get the content of a file by ID."""
	try:
	# Get the file metadata to determine the type
	file_metadata = drive_service.files().get(fileId=file_id, fields="name,mimeType").execute()
	mime_type = file_metadata['mimeType']
	file_name = file_metadata['name']

	# Handle different file types
	if mime_type == 'application/vnd.google-apps.spreadsheet':
	# For Google Sheets, get the data using Sheets API
	result = sheets_service.spreadsheets().values().get(
	spreadsheetId=file_id, range='A1:Z1000').execute()
	rows = result.get('values', [])

	if not rows:
	return f"No data found in the sheet: {file_name}"

	# Format as CSV
	csv_content = io.StringIO()
	for row in rows:
	csv_content.write(','.join([str(cell) for cell in row]) + '\n')

	return csv_content.getvalue()

	elif mime_type in ['text/plain', 'text/csv', 'application/json']:
	# For text files, download the content
	request = drive_service.files().get_media(fileId=file_id)
	file_content = io.BytesIO()
	downloader = MediaIoBaseDownload(file_content, request)

	done = False
	while not done:
	status, done = downloader.next_chunk()

	return file_content.getvalue().decode('utf-8')

	else:
	return f"File '{file_name}' has unsupported MIME type: {mime_type}. Only text, CSV, JSON, and Google Sheets are supported."

	except HttpError as error:
	return f"An error occurred retrieving file: {error}"


	@mcp.resource(
	uri_schema="gdrive://sheets/list",
	name="Google Sheets List",
	description="List all Google Sheets"
	)
	@require_auth
	async def list_sheets() -> str:
	"""Lists all Google Sheets from Drive."""
	try:
	results = drive_service.files().list(
	q="mimeType='application/vnd.google-apps.spreadsheet'",
	pageSize=50,
	fields="nextPageToken, files(id, name, webViewLink)"
	).execute()
	items = results.get('files', [])

	if not items:
	return "No Google Sheets found."

	sheets_list = []
	for item in items:
	sheets_list.append({
	'id': item['id'],
	'name': item['name'],
	'webViewLink': item.get('webViewLink', 'N/A')
	})

	return json.dumps(sheets_list, indent=2)

	except HttpError as error:
	return f"An error occurred listing sheets: {error}"


	# Tools
	@mcp.tool()
	@require_auth
	async def search_files(query: str) -> str:
	"""Search for files in Google Drive.

	Args:
	query: Search query text (e.g., "budget 2024" or "type:spreadsheet")
	"""
	try:
	results = drive_service.files().list(
	q=f"fullText contains '{query}'",
	pageSize=20,
	fields="nextPageToken, files(id, name, mimeType, webViewLink)"
	).execute()
	items = results.get('files', [])

	if not items:
	return f"No files found matching query: {query}"

	file_list = []
	for item in items:
	file_list.append({
	'id': item['id'],
	'name': item['name'],
	'mimeType': item['mimeType'],
	'webViewLink': item.get('webViewLink', 'N/A')
	})

	return json.dumps(file_list, indent=2)

	except HttpError as error:
	return f"An error occurred searching files: {error}"


	@mcp.tool()
	@require_auth
	async def read_sheet_range(spreadsheet_id: str, range_name: str = "A1:Z1000") -> str:
	"""Read data from a specific range in a Google Sheet.

	Args:
	spreadsheet_id: The ID of the Google Sheet
	range_name: The A1 notation of the range to read (default: A1:Z1000)
	"""
	try:
	result = sheets_service.spreadsheets().values().get(
	spreadsheetId=spreadsheet_id, range=range_name).execute()
	rows = result.get('values', [])

	if not rows:
	return f"No data found in range: {range_name}"

	# Try to convert to a cleaner tabular format
	try:
	df = pd.DataFrame(rows)
	# Use first row as headers if there are at least 2 rows
	if len(rows) > 1:
	df.columns = df.iloc[0]
	df = df.drop(0)

	return df.to_string(index=False)
	except Exception:
	# Fall back to simple formatting if pandas conversion fails
	formatted_rows = []
	for row in rows:
	formatted_rows.append('\t'.join([str(cell) for cell in row]))

	return '\n'.join(formatted_rows)

	except HttpError as error:
	return f"An error occurred reading sheet range: {error}"


	@mcp.tool()
	@require_auth
	async def write_to_sheet(spreadsheet_id: str, range_name: str, data: str) -> str:
	"""Write data to a Google Sheet.

	Args:
	spreadsheet_id: The ID of the Google Sheet
	range_name: The A1 notation of the range to write (e.g., "Sheet1!A1:B5")
	data: CSV-formatted data to write (each line is a row, values separated by commas)
	"""
	try:
	# Convert the CSV-formatted data to a list of lists
	lines = data.strip().split('\n')
	values = [line.split(',') for line in lines]

	body = {
	'values': values
	}

	result = sheets_service.spreadsheets().values().update(
	spreadsheetId=spreadsheet_id,
	range=range_name,
	valueInputOption='USER_ENTERED',
	body=body
	).execute()

	return f"Updated {result.get('updatedCells')} cells in {range_name}"

	except HttpError as error:
	return f"An error occurred writing to sheet: {error}"


	@mcp.tool()
	@require_auth
	async def create_new_sheet(title: str) -> str:
	"""Create a new Google Sheet.

	Args:
	title: Title for the new spreadsheet
	"""
	try:
	spreadsheet_body = {
	'properties': {
	'title': title
	}
	}

	spreadsheet = sheets_service.spreadsheets().create(
	body=spreadsheet_body).execute()

	# Get the spreadsheet ID and URL
	spreadsheet_id = spreadsheet.get('spreadsheetId')
	spreadsheet_url = spreadsheet.get('spreadsheetUrl')

	return json.dumps({
	"message": f"Created new spreadsheet: {title}",
	"id": spreadsheet_id,
	"url": spreadsheet_url
	}, indent=2)

	except HttpError as error:
	return f"An error occurred creating the spreadsheet: {error}"


	@mcp.tool()
	@require_auth
	async def run_sheet_query(spreadsheet_id: str, sheet_name: str, query: str) -> str:
	"""Run a simplified SQL-like query on a Google Sheet.

	Args:
	spreadsheet_id: The ID of the Google Sheet
	sheet_name: The name of the sheet to query
	query: SQL-like query (supports: SELECT, WHERE, GROUP BY, ORDER BY, LIMIT)
	"""
	try:
	# Get the sheet data first
	result = sheets_service.spreadsheets().values().get(
	spreadsheetId=spreadsheet_id, range=f"{sheet_name}").execute()
	rows = result.get('values', [])

	if not rows:
	return f"No data found in sheet: {sheet_name}"

	# Convert to pandas DataFrame
	df = pd.DataFrame(rows[1:], columns=rows[0])

	# Interpret the query
	if not query.lower().startswith('select'):
	query = 'SELECT * ' + query

	# Handle basic SQL operations
	# This is a very simplified SQL parser for demo purposes
	query_parts = query.lower().split()

	# Extract where clause if it exists
	where_clause = None
	if 'where' in query_parts:
	where_index = query_parts.index('where')
	# Find the next clause after WHERE
	next_clause_keywords = ['group', 'order', 'limit']
	next_clause_indices = [query_parts.index(kw) for kw in next_clause_keywords if kw in query_parts]

	if next_clause_indices:
	end_index = min(next_clause_indices)
	where_clause = ' '.join(query_parts[where_index+1:end_index])
	else:
	where_clause = ' '.join(query_parts[where_index+1:])

	# Extract order by clause
	order_by = None
	if 'order' in query_parts and 'by' in query_parts:
	order_index = query_parts.index('order')
	if query_parts[order_index + 1] == 'by':
	# Same logic to find the end of the ORDER BY clause
	next_clause_keywords = ['limit']
	next_clause_indices = [query_parts.index(kw) for kw in next_clause_keywords if kw in query_parts]

	if next_clause_indices:
	end_index = min(next_clause_indices)
	order_by = ' '.join(query_parts[order_index+2:end_index])
	else:
	order_by = ' '.join(query_parts[order_index+2:])

	# Extract limit
	limit = None
	if 'limit' in query_parts:
	limit_index = query_parts.index('limit')
	if limit_index + 1 < len(query_parts):
	try:
	limit = int(query_parts[limit_index + 1])
	except ValueError:
	pass

	# Apply filters through Python code
	# Note: This is a very basic implementation
	if where_clause:
	# Here we'd need a more sophisticated parser for real SQL WHERE clauses
	# This is just a simple example that handles basic conditions
	try:
	# For simple conditions like "column = value"
	parts = where_clause.split('=')
	if len(parts) == 2:
	col = parts[0].strip()
	val = parts[1].strip().strip("'\"")
	df = df[df[col] == val]
	except Exception as e:
	return f"Error parsing WHERE clause: {e}"

	# Apply ordering
	if order_by:
	try:
	# Simple order by implementation
	parts = order_by.split()
	col = parts[0].strip()
	ascending = True
	if len(parts) > 1 and parts[1].upper() == 'DESC':
	ascending = False
	df = df.sort_values(by=col, ascending=ascending)
	except Exception as e:
	return f"Error applying ORDER BY: {e}"

	# Apply limit
	if limit:
	df = df.head(limit)

	# Return the result
	return df.to_string(index=False)

	except HttpError as error:
	return f"An error occurred querying the sheet: {error}"


	@mcp.tool()
	@require_auth
	async def get_file_metadata(file_id: str) -> str:
	"""Get detailed metadata for a specific file by ID.

	Args:
	file_id: The ID of the file in Google Drive
	"""
	try:
	file = drive_service.files().get(
	fileId=file_id,
	fields="id, name, mimeType, description, createdTime, modifiedTime, webViewLink, size"
	).execute()

	return json.dumps(file, indent=2)

	except HttpError as error:
	return f"An error occurred getting file metadata: {error}"


	# Prompts
	@mcp.prompt()
	@require_auth
	async def analyze_sheet(sheet_id: str, analysis_type: str = "summary") -> List[Dict[str, Any]]:
	"""Prompt for analyzing Google Sheet data.

	Args:
	sheet_id: ID of the Google Sheet to analyze
	analysis_type: Type of analysis (summary, trends, outliers)
	"""
	result = sheets_service.spreadsheets().values().get(
	spreadsheetId=sheet_id, range="A1:Z1000").execute()
	rows = result.get('values', [])

	if not rows:
	return [
	{
	"role": "user",
	"content": f"The Google Sheet with ID '{sheet_id}' is empty. Please provide data before analysis."
	}
	]

	# Get the metadata for the sheet name
	sheet_metadata = drive_service.files().get(fileId=sheet_id, fields="name").execute()
	sheet_name = sheet_metadata.get('name', 'Unnamed Sheet')

	# Create a DataFrame for analysis
	df = pd.DataFrame(rows[1:], columns=rows[0])

	# Generate different prompts based on analysis type
	if analysis_type.lower() == "summary":
	return [
	{
	"role": "user",
	"content": f"""
	Analyze this Google Sheet data from '{sheet_name}':

	Column Headers: {', '.join(df.columns.tolist())}

	Number of Rows: {len(df)}

	Sample Data (first 5 rows):
	{df.head(5).to_string(index=False)}

	Please provide:
	1. A summary of what this data represents
	2. Key statistics for numeric columns
	3. Distribution of categorical data
	4. Any notable patterns
	5. Recommendations for further analysis
	"""
	}
	]

	elif analysis_type.lower() == "trends":
	return [
	{
	"role": "user",
	"content": f"""
	Analyze trends in this Google Sheet data from '{sheet_name}':

	Column Headers: {', '.join(df.columns.tolist())}

	Sample Data (first 5 rows):
	{df.head(5).to_string(index=False)}

	Please identify:
	1. Temporal trends (if date/time columns exist)
	2. Correlation between numeric columns
	3. Changes in distributions over any sequential variables
	4. Growth rates or decline patterns
	5. Cyclical patterns if they exist
	6. Recommendations for visualizing these trends
	"""
	}
	]

	elif analysis_type.lower() == "outliers":
	return [
	{
	"role": "user",
	"content": f"""
	Analyze outliers in this Google Sheet data from '{sheet_name}':

	Column Headers: {', '.join(df.columns.tolist())}

	Please identify:
	1. Statistical outliers in numeric columns
	2. Unusual patterns or combinations in the data
	3. Potential data entry errors
	4. Impact of outliers on overall analysis
	5. Recommendations for handling these outliers

	Here's a statistical summary of numeric columns:
	{df.describe().to_string()}
	"""
	}
	]

	else:
	return [
	{
	"role": "user",
	"content": f"""
	Analyze this Google Sheet data from '{sheet_name}' based on your judgment:

	Column Headers: {', '.join(df.columns.tolist())}

	Sample Data (first 10 rows):
	{df.head(10).to_string(index=False)}

	Please provide a comprehensive analysis including key insights and recommendations.
	"""
	}
	]


	@mcp.prompt()
	@require_auth
	async def create_report_template(report_type: str) -> List[Dict[str, Any]]:
	"""Prompt for creating a Google Sheets report template.

	Args:
	report_type: Type of report (financial, project_tracker, dashboard, inventory)
	"""
	report_prompts = {
	"financial": """
	Create a comprehensive financial report template in Google Sheets with the following components:

	1. Income Statement sheet with:
	- Revenue categories
	- Expense categories
	- Profit calculations
	- Month-to-month comparisons
	- Year-to-date totals

	2. Balance Sheet with:
	- Assets section
	- Liabilities section
	- Equity section
	- Auto-calculated totals

	3. Cash Flow Statement with:
	- Operating activities
	- Investing activities
	- Financing activities

	4. Financial Ratios dashboard with:
	- Liquidity ratios
	- Profitability ratios
	- Leverage ratios
	- Efficiency ratios

	Include formulas, conditional formatting, and data validation where appropriate.
	Explain the key components and how to use them effectively.
	""",

	"project_tracker": """
	Design a comprehensive project management tracker in Google Sheets with the following components:

	1. Project Overview dashboard with:
	- Project name, description, objectives
	- Key dates and milestones
	- Budget summary
	- Overall status indicators

	2. Task Tracker sheet with:
	- Task descriptions
	- Assignees
	- Due dates
	- Status (Not Started, In Progress, Completed, Delayed)
	- Priority levels
	- Dependencies
	- Progress percentage
	- Notes/comments

	3. Resource Allocation sheet with:
	- Team members
	- Hours allocation
	- Skills matrix
	- Availability calendar

	4. Budget Tracker with:
	- Planned vs actual expenses
	- Cost categories
	- Variance calculations

	5. Risk Register with:
	- Risk descriptions
	- Impact and probability ratings
	- Mitigation strategies
	- Status tracking

	Include formulas for automatic calculations, conditional formatting for visual cues,
	and data validation to maintain data integrity. Explain how each component works and
	suggest best practices for using this tracker effectively.
	""",

	"dashboard": """
	Create a versatile business KPI dashboard template in Google Sheets with the following components:

	1. Executive Summary with:
	- Key performance indicators at a glance
	- Performance vs targets visualization
	- Period-over-period comparisons
	- Traffic light indicators for status

	2. Sales Performance section with:
	- Revenue by product/service
	- Sales by region/channel
	- Conversion metrics
	- Sales pipeline visualization

	3. Marketing Metrics section with:
	- Campaign performance
	- Customer acquisition cost
	- Channel effectiveness
	- Engagement metrics

	4. Customer Insights section with:
	- Customer satisfaction scores
	- Retention metrics
	- Lifetime value calculations
	- Churn analysis

	5. Operations section with:
	- Production/service delivery metrics
	- Quality indicators
	- Efficiency measures
	- Inventory levels (if applicable)

	6. Financial Overview with:
	- Profit margins
	- Cash flow status
	- Expense breakdown
	- Financial forecasts

	Include instructions for data input, formulas for calculations, and guidance on creating
	visualizations like charts and sparklines. Design the dashboard for clarity and ease of interpretation
	with conditional formatting and appropriate data presentation techniques.
	""",

	"inventory": """
	Design a comprehensive inventory management system in Google Sheets with the following components:

	1. Inventory Dashboard with:
	- Total inventory value
	- Stock level summary
	- Low stock alerts
	- Inventory turnover metrics
	- Value by category/location

	2. Master Inventory List with:
	- Item IDs and descriptions
	- Categories and subcategories
	- Quantities on hand
	- Reorder points
	- Cost and retail pricing
	- Supplier information
	- Storage location
	- Last count date

	3. Inventory Transactions Log with:
	- Date and time of transactions
	- Transaction types (receiving, sales, adjustments, transfers)
	- Quantity changes
	- User tracking
	- Notes/reasons

	4. Supplier Management section with:
	- Supplier details
	- Order history
	- Performance metrics
	- Contact information

	5. Reports and Analytics section with:
	- Sales velocity by product
	- Dead stock identification
	- Seasonal trends
	- Category performance
	- Forecasting tools

	Include formulas for automatic calculations (like FIFO/LIFO valuation), data validation for entry,
	conditional formatting for visual alerts, and instructions on implementing barcode scanning
	if applicable. Explain how each component works together and best practices for maintaining
	accurate inventory records.
	"""
	}

	# Default generic report if type not found
	default_prompt = """
	Create a professional report template in Google Sheets with:

	1. An executive summary dashboard
	2. Detailed data tables
	3. Analysis sections
	4. Visualizations and charts
	5. Conclusions and recommendations

	Include appropriate formatting, formulas, and instructions for users to customize for their needs.
	"""

	# Get the specific prompt or fallback to default
	prompt_content = report_prompts.get(report_type.lower(), default_prompt)

	return [
	{
	"role": "user",
	"content": prompt_content.strip()
	}
	]


	def parse_args():
	"""Parse command line arguments."""
	parser = argparse.ArgumentParser(description='Google Drive MCP Server')
	parser.add_argument('--stdio', action='store_true', help='Use stdio transport')
	return parser.parse_args()


	if __name__ == "__main__":
	args = parse_args()

	# Start the server with specified transport
	if args.stdio:
	mcp.run(transport='stdio')
	else:
	# Default to stdio if no transport specified
	mcp.run(transport='stdio')
No results found