MCP SQL Server

""" Example: Using execute_select_query tool to safely execute SELECT queries This example demonstrates how to use the execute_select_query tool to: 1. Execute safe SELECT queries 2. Handle query validation and security 3. Process query results Security features demonstrated: - Blocking of dangerous keywords (UPDATE, DELETE, INSERT, etc.) - Comment stripping to prevent SQL injection - Row limit enforcement """ import asyncio from mcp_sql import MCPSQLServer from fastmcp import Context async def example_queries(server: MCPSQLServer): """Run example queries demonstrating the execute_select_query tool.""" # Get the tool instance execute_select = None for tool in server.tools: if tool.name == "execute_select_query": execute_select = tool break if not execute_select: print("❌ execute_select_query tool not found!") return print("=" * 80) print("🧪 EXECUTE SELECT QUERY TOOL - EXAMPLES") print("=" * 80) # Create a dummy context (in real usage, this comes from MCP) ctx = Context() # Database credentials (replace with your actual credentials) db_config = { "server_name": "localhost", "database": "test_db", "user": "test_user", "password": "test_password", } # ======================================================================== # EXAMPLE 1: Valid SELECT query # ======================================================================== print("\n📋 Example 1: Valid SELECT query") print("-" * 80) query1 = "SELECT TOP 10 id, name, email FROM users WHERE active = 1" print(f"Query: {query1}") result1 = await execute_select.execute( ctx=ctx, query=query1, limit=10, **db_config ) print(f"Result: {result1}") # ======================================================================== # EXAMPLE 2: SELECT with JOIN # ======================================================================== print("\n📋 Example 2: SELECT with JOIN") print("-" * 80) query2 = """ SELECT u.id, u.name, COUNT(o.id) as order_count, SUM(o.total) as total_spent FROM users u LEFT JOIN orders o ON u.id = o.user_id GROUP BY u.id, u.name ORDER BY total_spent DESC """ print(f"Query: {query2.strip()}") result2 = await execute_select.execute( ctx=ctx, query=query2, limit=20, **db_config ) print(f"Result: {result2}") # ======================================================================== # EXAMPLE 3: Blocked query - UPDATE attempt # ======================================================================== print("\n🚫 Example 3: Blocked query - UPDATE attempt") print("-" * 80) query3 = "SELECT * FROM users WHERE id = 1; UPDATE users SET status='admin' WHERE id = 1" print(f"Query: {query3}") result3 = await execute_select.execute( ctx=ctx, query=query3, **db_config ) print(f"Result: {result3}") print("✅ Security validation working - UPDATE blocked!") # ======================================================================== # EXAMPLE 4: Blocked query - DELETE attempt # ======================================================================== print("\n🚫 Example 4: Blocked query - DELETE attempt") print("-" * 80) query4 = "SELECT * FROM logs WHERE date < '2020-01-01'; DELETE FROM logs WHERE date < '2020-01-01'" print(f"Query: {query4}") result4 = await execute_select.execute( ctx=ctx, query=query4, **db_config ) print(f"Result: {result4}") print("✅ Security validation working - DELETE blocked!") # ======================================================================== # EXAMPLE 5: Blocked query - CREATE attempt hidden in comment # ======================================================================== print("\n🚫 Example 5: Blocked query - CREATE attempt hidden in comment") print("-" * 80) query5 = "/* SELECT * FROM temp */ CREATE TABLE backdoor (id INT)" print(f"Query: {query5}") result5 = await execute_select.execute( ctx=ctx, query=query5, **db_config ) print(f"Result: {result5}") print("✅ Security validation working - CREATE blocked even with comments!") # ======================================================================== # EXAMPLE 6: Blocked query - EXEC attempt # ======================================================================== print("\n🚫 Example 6: Blocked query - EXEC attempt") print("-" * 80) query6 = "SELECT * FROM users; EXEC sp_executesql 'DROP TABLE users'" print(f"Query: {query6}") result6 = await execute_select.execute( ctx=ctx, query=query6, **db_config ) print(f"Result: {result6}") print("✅ Security validation working - EXEC blocked!") # ======================================================================== # EXAMPLE 7: Blocked query - Not starting with SELECT # ======================================================================== print("\n🚫 Example 7: Blocked query - Not starting with SELECT") print("-" * 80) query7 = "-- This is a comment\nINSERT INTO users VALUES (1, 'hacker')" print(f"Query: {query7}") result7 = await execute_select.execute( ctx=ctx, query=query7, **db_config ) print(f"Result: {result7}") print("✅ Security validation working - Must start with SELECT!") # ======================================================================== # EXAMPLE 8: Valid complex query with subquery # ======================================================================== print("\n📋 Example 8: Valid complex query with subquery") print("-" * 80) query8 = """ SELECT department, avg_salary, employee_count FROM ( SELECT department, AVG(salary) as avg_salary, COUNT(*) as employee_count FROM employees WHERE hire_date > '2020-01-01' GROUP BY department ) dept_stats WHERE employee_count > 5 ORDER BY avg_salary DESC """ print(f"Query: {query8.strip()}") result8 = await execute_select.execute( ctx=ctx, query=query8, limit=50, **db_config ) print(f"Result: {result8}") print("\n" + "=" * 80) print("✅ All examples completed!") print("=" * 80) def print_result_formatted(result: dict): """Pretty print query results.""" if "error" in result: print(f"❌ Error: {result['error']}") return print(f"✅ Success!") print(f" Columns: {', '.join(result['columns'])}") print(f" Rows returned: {result['row_count']}") if result['rows']: print(f"\n Sample data (first 3 rows):") for i, row in enumerate(result['rows'][:3], 1): print(f" {i}. {row}") def main(): """Main entry point.""" print("🚀 Starting MCP SQL Server with execute_select_query tool...") # Create server instance server = MCPSQLServer() print("\n📦 Registered tools:") for tool in server.tools: print(f" • {tool.name}: {tool.description}") # Run examples print("\n🧪 Running query examples...") asyncio.run(example_queries(server)) print("\n" + "=" * 80) print("💡 TIPS:") print("=" * 80) print("1. Always use SELECT for read-only operations") print("2. Specify reasonable LIMIT values to avoid performance issues") print("3. Use WHERE clauses to filter data efficiently") print("4. The tool automatically blocks dangerous operations (UPDATE, DELETE, etc.)") print("5. Comments in SQL are stripped before validation for security") print("6. Maximum row limit is 1000 (hard cap for safety)") print("=" * 80) if __name__ == "__main__": main()