Google Workspace MCP Server

#!/usr/bin/env python3 """ PostgreSQL MCP Server for Claude Desktop Provides database operations via Model Context Protocol """ import sys import json import asyncio from pathlib import Path sys.path.insert(0, str(Path(__file__).parent.parent)) from mcp.server import Server from mcp.types import Tool, TextContent from mcp.server.stdio import stdio_server import src.postgres_db as db def load_config(): """Load database configuration""" try: # Try multiple possible paths possible_paths = [ Path(__file__).parent.parent / 'config.json', # Standard location Path.cwd() / 'config.json', # Current working directory Path.home() / 'mcplatestv1' / 'config.json', # Alternative location ] for config_path in possible_paths: if config_path.exists(): with open(config_path, 'r') as f: return json.load(f) # If no config file found, return empty dict return {'database': {}} except Exception as e: # Return error info for debugging return {'database': {}, 'error': str(e)} config = load_config() db_url = config.get('database', {}).get('url', '') server = Server("postgresql-mcp-server") @server.list_tools() async def list_tools(): """List all available database tools""" return [ Tool( name="execute_query", description="Execute a SELECT query and return results. Use for reading data from tables.", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "SQL SELECT query to execute" }, "params": { "type": "array", "description": "Optional query parameters for parameterized queries", "items": {"type": "string"} } }, "required": ["query"] } ), Tool( name="execute_write", description="Execute INSERT, UPDATE, or DELETE queries. Use for modifying data.", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "SQL INSERT, UPDATE, or DELETE query" }, "params": { "type": "array", "description": "Optional query parameters", "items": {"type": "string"} } }, "required": ["query"] } ), Tool( name="run_custom_sql", description="Execute any SQL query (SELECT, INSERT, UPDATE, DELETE, etc.). Automatically handles query type.", inputSchema={ "type": "object", "properties": { "sql": { "type": "string", "description": "Any SQL query to execute" }, "params": { "type": "array", "description": "Optional query parameters", "items": {"type": "string"} } }, "required": ["sql"] } ), Tool( name="list_tables", description="List all tables in the database with their schemas", inputSchema={ "type": "object", "properties": {} } ), Tool( name="describe_table", description="Get detailed schema information for a specific table including columns, types, and constraints", inputSchema={ "type": "object", "properties": { "table_name": { "type": "string", "description": "Name of the table to describe" } }, "required": ["table_name"] } ), Tool( name="get_table_count", description="Get the total number of rows in a table", inputSchema={ "type": "object", "properties": { "table_name": { "type": "string", "description": "Name of the table" } }, "required": ["table_name"] } ) ] @server.call_tool() async def call_tool(name: str, arguments: dict): """Handle tool execution""" try: # Reload config on each call to ensure we have the latest current_config = load_config() current_db_url = current_config.get('database', {}).get('url', '') if not current_db_url: # Try to provide helpful error message config_path = Path(__file__).parent.parent / 'config.json' error_msg = { "success": False, "error": "Database URL not configured", "details": "Please add 'database.url' to config.json", "config_location": str(config_path), "config_exists": config_path.exists(), "config_keys": list(current_config.keys()) } if 'error' in current_config: error_msg["config_error"] = current_config['error'] return [TextContent(type="text", text=json.dumps(error_msg, indent=2))] result = None if name == "execute_query": query = arguments.get("query") params = arguments.get("params") result = db.execute_query(current_db_url, query, params) elif name == "execute_write": query = arguments.get("query") params = arguments.get("params") result = db.execute_write(current_db_url, query, params) elif name == "run_custom_sql": sql = arguments.get("sql") params = arguments.get("params") result = db.run_custom_sql(current_db_url, sql, params) elif name == "list_tables": result = db.list_tables(current_db_url) elif name == "describe_table": table_name = arguments.get("table_name") result = db.describe_table(current_db_url, table_name) elif name == "get_table_count": table_name = arguments.get("table_name") result = db.get_table_count(current_db_url, table_name) else: result = {"success": False, "error": f"Unknown tool: {name}"} return [TextContent(type="text", text=json.dumps(result, indent=2, default=str))] except Exception as error: return [TextContent( type="text", text=json.dumps({"success": False, "error": str(error)}, indent=2) )] async def main(): """Main entry point for MCP server""" async with stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, server.create_initialization_options() ) if __name__ == "__main__": asyncio.run(main())