SQL Server MCP

Instructions

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

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}