SQL Server MCP

Instructions

Implementation Reference

• src/mssql_mcp/tools/databases.py:83-149 (handler)

  The main handler function for the 'switch_database' tool. It is decorated with @mcp.tool(), which registers it as an MCP tool. The function validates the database name, checks if it's blocked or exists and online, switches using 'USE [db]' statement, and returns status with previous and new database names.
  @mcp.tool() def switch_database(database_name: str) -> dict[str, Any]: """Switch the active database context. Changes the current database using the USE statement. The database must exist, be online, and not be in the blocklist (MSSQL_BLOCKED_DATABASES). Args: database_name: Name of the database to switch to Returns: Dictionary with: - status: "switched" on success, "error" on failure - database: The new active database name - previous_database: The previously active database - error: Error message if switch failed """ try: manager = get_connection_manager() config = manager.config blocked_databases = config.blocked_databases # Check if database is blocked if database_name.lower() in blocked_databases: return { "status": "error", "error": f"Access to database '{database_name}' is not allowed", "database": database_name, } # Get current database before switching current_db_query = "SELECT DB_NAME() AS current_database" current_db_result = manager.execute_query(current_db_query) previous_database = ( current_db_result[0]["current_database"] if current_db_result else None ) # Switch database using USE statement # Note: USE cannot be parameterized, but we validate the name exists first # by checking sys.databases check_query = "SELECT name FROM sys.databases WHERE name = %s AND state_desc = 'ONLINE'" check_result = manager.execute_query(check_query, (database_name,)) if not check_result: return { "status": "error", "error": f"Database '{database_name}' does not exist or is not online", "database": database_name, } # Execute USE statement (database name is validated, use bracket quoting for safety) use_query = f"USE [{database_name}]" manager.execute_query(use_query) # Verify the switch verify_result = manager.execute_query(current_db_query) new_database = verify_result[0]["current_database"] if verify_result else None return { "status": "switched", "database": new_database, "previous_database": previous_database, } except Exception as e: logger.error(f"Error switching database: {e}") return {"error": str(e), "database": database_name}